DuckDB ASOF Join 實戰：時間序列對齊與最近事件查詢

Python 學習 - 本文屬於一個選集。

§ 124: 本文

一. 前言：兩張時間表，為什麼總是對不齊？
#

想像一個很常見的資料工程場景：

events 每幾秒記錄一次裝置讀值；
settings 只在設定變更時留下一筆；
分析時，要知道每個 event 當下沿用哪一版設定。

兩張表的時間幾乎不可能完全相等。如果用普通的等值 JOIN，大部分 event 都配不到資料；如果先做笛卡兒積再找最接近時間，SQL 又長又容易把資料量炸開。 DuckDB 的 ASOF JOIN 就是為這種「截至這個時間，最近有效的是哪一筆？」而設計。今天拍拍君不重講 DuckDB 如何讀 CSV 或 Parquet。如果你還沒用過 DuckDB，可以先看 Python DuckDB 入門；如果需求是排名、移動平均或 LAG / LEAD，則看 DuckDB Window Functions。這篇只專心處理一件事：把不規則時間序列安全地對齊。

二. 安裝與範例目標
#

用 uv 建一個最小環境：

uv init duckdb-asof-demo
cd duckdb-asof-demo
uv add "duckdb==1.5.5" pandas pyarrow

也可以用 pip：

python -m pip install "duckdb==1.5.5" pandas pyarrow

本文會把裝置事件配到「同一台裝置、且不晚於事件時間」的最近設定。先建立資料：

import duckdb
con = duckdb.connect()
con.execute("""
CREATE TABLE events (
    event_id INTEGER,
    device_id VARCHAR,
    observed_at TIMESTAMP,
    temperature DOUBLE
);
INSERT INTO events VALUES
    (1, 'pypy-01', '2026-09-23 10:00:05', 21.4),
    (2, 'pypy-01', '2026-09-23 10:03:20', 23.1),
    (3, 'pypy-01', '2026-09-23 10:12:00', 24.8),
    (4, 'pypy-02', '2026-09-23 10:02:00', 19.7),
    (5, 'pypy-03', '2026-09-23 10:05:00', 20.2);
CREATE TABLE settings (
    setting_id INTEGER,
    device_id VARCHAR,
    valid_from TIMESTAMP,
    target_temperature DOUBLE
);
INSERT INTO settings VALUES
    (101, 'pypy-01', '2026-09-23 09:55:00', 22.0),
    (102, 'pypy-01', '2026-09-23 10:02:00', 24.0),
    (201, 'pypy-02', '2026-09-23 10:00:00', 20.0);
""")

注意兩個時間欄位的語意：

events.observed_at：問題發生的時間；
settings.valid_from：右表狀態開始生效的時間。

欄位名稱不是重點，時間代表的事件語意才是重點。

三. 第一個 ASOF JOIN：找最近的過去狀態
#

核心查詢只有幾行：

SELECT
    e.event_id,
    e.device_id,
    e.observed_at,
    e.temperature,
    s.setting_id,
    s.valid_from,
    s.target_temperature
FROM events AS e
ASOF LEFT JOIN settings AS s
    ON e.device_id = s.device_id
   AND e.observed_at >= s.valid_from
ORDER BY e.event_id;

這個條件可以讀成：

對每個 event，找同一台 device 中，valid_from 不晚於 observed_at 的最近一筆 setting。 ASOF 對每個左表 row 最多只選一筆右表資料。它不是把所有歷史設定都展開，因此很適合狀態表、行情、費率、模型版本與設定版本。預期配對如下：

event	event time	matched setting	setting time
1	10:00:05	101	09:55:00
2	10:03:20	102	10:02:00
3	10:12:00	102	10:02:00
4	10:02:00	201	10:00:00
5	10:05:00	NULL	NULL

pypy-03 完全沒有設定，但因為使用 ASOF LEFT JOIN，event 還是會保留。

四. 左右順序與不等式方向，真的不能亂放
#

普通等值 join 常讓人覺得左右交換只是欄位順序不同。 ASOF JOIN 不是這樣。這個版本找「最近的過去」：

e.observed_at >= s.valid_from

左表是要被補資料的 event；右表是可供查找的狀態歷史。若改成：

e.observed_at <= s.valid_from

語意就變成找 event 之後最近的一筆 setting，也就是「最近的未來」。這有時適合配對下一個排程、下一個到站時間或下一個截止點，但若拿來重建 event 當下狀態，就會造成資料穿越。先把需求寫成一句人話，再選不等式：

問題	條件	方向
當時最新的狀態	`left_time >= right_time`	backward
接下來第一個事件	`left_time <= right_time`	forward

DuckDB 的 ASOF 排序欄位也可以用 > 或 <。嚴格不等式會排除時間完全相同的 row；除非規格真的如此，通常先用 >= 或 <=。

五. 分組鍵：最近，但必須是同一個世界
#

只比時間通常不夠。裝置 A 的 event 不可以拿到裝置 B 的設定，股票 A 也不能拿到股票 B 的價格。因此 ASOF 條件通常由兩類 predicate 組成：

ON e.device_id = s.device_id
AND e.observed_at >= s.valid_from

一個 ordering inequality：決定時間方向與最近值；
零個或多個 equality：限制 partition。

例如多租戶系統應把 tenant 一起加入：

ON e.tenant_id = s.tenant_id
AND e.device_id = s.device_id
AND e.observed_at >= s.valid_from

少一個 equality 不一定報錯，卻可能得到看似合理的錯資料。這種 bug 最討厭，因為 row 數、型別與時間都可能正常。如果 key 允許 NULL，要先定義 NULL 是否代表同一組。 DuckDB 允許其他條件使用 equality 或 NOT DISTINCT，但實務上更建議先清理 key，避免把「未知裝置」誤當成一個共享 partition。

六. 排序規則：不是先 `ORDER BY` 就萬事大吉
#

ASOF 的「ordered」主要來自不等式欄位的語意。你不需要為正確性先把兩張 DuckDB table 手動排序；查詢引擎會執行需要的工作。但有三件事仍要明確處理：

輸出沒有保證順序，需要展示或測試時仍要 ORDER BY；
左右時間型別應相容，別讓字串排序冒充時間排序；
同一 partition、同一時間若有多筆右表資料，必須先定義勝負。

先檢查型別：

DESCRIBE events;
DESCRIBE settings;

CSV 若把時間讀成 VARCHAR，應在資料邊界轉成 TIMESTAMP：

SELECT
    device_id,
    try_cast(valid_from AS TIMESTAMP) AS valid_from,
    target_temperature
FROM read_csv('settings.csv');

不要仰賴 2026-9-3 這類不固定格式字串的字典序。

七. 右表同秒重複：先決定誰才是正式版本
#

假設同一台裝置在同一時間留下兩個設定：

INSERT INTO settings VALUES
    (103, 'pypy-01', '2026-09-23 10:02:00', 23.5);

只靠 valid_from 無法表達 102 與 103 哪筆優先。不要期待資料庫替你猜。可以在 join 前用 window function 建立 deterministic right side：

WITH canonical_settings AS (
    SELECT * EXCLUDE (revision_rank)
    FROM (
        SELECT
            *,
            row_number() OVER (
                PARTITION BY device_id, valid_from
                ORDER BY setting_id DESC
            ) AS revision_rank
        FROM settings
    )
    WHERE revision_rank = 1
)
SELECT
    e.event_id,
    e.observed_at,
    s.setting_id,
    s.valid_from,
    s.target_temperature
FROM events AS e
ASOF LEFT JOIN canonical_settings AS s
    ON e.device_id = s.device_id
   AND e.observed_at >= s.valid_from
ORDER BY e.event_id;

這裡把較大的 setting_id 當成較新的修訂。正式系統可能改用 revision、ingested_at 或事件序號。重點是 tie-break 規則必須來自資料契約，而不是碰巧的檔案順序。

八. Tolerance：最近，不代表夠近
#

第三筆 event 在 10:12，最近設定是 10:02。 SQL 的確找到最近值，但 10 分鐘前的設定是否仍可信，要看業務規則。先保留 match age：

WITH matched AS (
    SELECT
        e.*,
        s.setting_id,
        s.valid_from,
        s.target_temperature,
        e.observed_at - s.valid_from AS match_age
    FROM events AS e
    ASOF LEFT JOIN settings AS s
        ON e.device_id = s.device_id
       AND e.observed_at >= s.valid_from
)
SELECT
    *,
    match_age <= INTERVAL '5 minutes' AS within_tolerance
FROM matched
ORDER BY event_id;

ASOF 的 join condition 需要一個 ordering inequality；其他 join 條件必須是 equality（或 NOT DISTINCT）。因此不要硬把另一個範圍 inequality 當成第二條 ASOF 時間條件。更穩定的做法是：

先找最近候選；
再把 match_age 當成品質與採用政策。

如果超過 5 分鐘要視為未配對，同時保留左表 row：

WITH matched AS (
    SELECT
        e.*,
        s.setting_id,
        s.valid_from,
        s.target_temperature,
        e.observed_at - s.valid_from AS match_age
    FROM events AS e
    ASOF LEFT JOIN settings AS s
        ON e.device_id = s.device_id
       AND e.observed_at >= s.valid_from
)
SELECT
    event_id,
    device_id,
    observed_at,
    temperature,
    CASE WHEN match_age <= INTERVAL '5 minutes' THEN setting_id END AS setting_id,
    CASE WHEN match_age <= INTERVAL '5 minutes' THEN valid_from END AS valid_from,
    CASE WHEN match_age <= INTERVAL '5 minutes' THEN target_temperature END AS target_temperature,
    match_age
FROM matched
ORDER BY event_id;

為什麼不直接在外層 WHERE match_age <= ...？因為那會把超時 event 整列刪掉，破壞 left join 想保留左表的目的。

九. 從 Python 執行：參數只控制政策，不拼 SQL
#

把 tolerance 做成參數：

from datetime import timedelta
query = """
WITH matched AS (
    SELECT
        e.event_id,
        e.device_id,
        e.observed_at,
        s.setting_id,
        s.valid_from,
        e.observed_at - s.valid_from AS match_age
    FROM events AS e
    ASOF LEFT JOIN settings AS s
        ON e.device_id = s.device_id
       AND e.observed_at >= s.valid_from
)
SELECT
    *,
    coalesce(match_age <= $max_age, false) AS accepted
FROM matched
ORDER BY event_id
"""
result = con.execute(
    query,
    {"max_age": timedelta(minutes=5)},
).df()
print(result)

參數值交給 DuckDB 處理，不要用 f-string 拼使用者輸入。若資料已在 pandas DataFrame，也可以讓 DuckDB replacement scan 直接讀取，但正式管線最好替資料來源取明確名稱並集中 connection 生命週期。

十. Temporal QA：別只驗證 SQL 有沒有跑完
#

時間 join 最危險的不是 query 失敗，而是 query 成功卻語意錯誤。先把正式查詢存成 matched_events view 或 table，再執行這組檢查：

SELECT count(*) FROM events;
SELECT count(*) FROM matched_events;
SELECT device_id, count(*) AS unmatched_rows
FROM matched_events
WHERE setting_id IS NULL
GROUP BY device_id;
SELECT count(*) AS future_leaks
FROM matched_events
WHERE valid_from > observed_at;
SELECT
    min(match_age),
    median(match_age),
    quantile_cont(match_age, 0.95),
    max(match_age)
FROM matched_events;

要驗證的契約是：

ASOF LEFT JOIN 前後的左表 row 數相同；
backward join 的 future_leaks 必須是 0；
unmatched 應能用新裝置或資料缺口解釋；
p95 與最大 match_age 沒有超出業務容忍值；
(partition key, right time) 重複時已套用 revision 規則。

測試資料至少放入同秒配對、早於第一筆、晚於最後一筆、partition 不存在、右表同秒多 revision，以及 tolerance 剛好相等。只測快樂路徑，等於沒測時間邊界。效能方面可用 EXPLAIN ANALYZE 看實際計畫，並遵守幾個原則：只 select 需要欄位、先用日期或 partition pruning 限制掃描、讓時間型別一致、先 canonicalize 右表重複 revision。裁切歷史時要保留 lookback；早上第一筆 event 可能仍需要昨晚最後一筆設定。最後，ASOF 回答的是「單一最近候選」。要找時間窗內所有事件請用 range join；要比較前後 row 請用 LAG / LEAD；要做固定分鐘聚合請用 time bucket；要線性插值則要同時找前後兩側並寫清楚公式。