Log Module ZH

English version

日誌模組 — beam_pipeline_toolkit/log/

適用於任何以此工具箱建構之管線的通用、選用日誌記錄——與 pipeline/ 的商業邏輯掛勾點及 ecommerce/ 完全獨立。

execution_log.py

回答的問題是：「某個特定階段內部是否發生了非預期事件，是哪一種？」

• exception_event(schema, run_id, stage, exc) — 為捕捉到的例外建立消毒過的事件，只揭露例外的類型名稱，絕不使用 str(exc)(它可能內嵌觸發例外的紀錄本身)。呼叫端仍會在之後重新拋出例外——不影響 runner 本身的重試/失敗機制。
• build_execution_log_event(...) — 較底層的建構函式；會依 schema 驗證 stage，並驗證 severity 是否屬於 {INFO, WARNING, ERROR}。
• TAG_EXECUTION_LOG — 每個 DoFn 用來註冊此事件串流的 TaggedOutput 標籤。

stage_timing.py

回答的問題是：「每個階段各花了多少時間？」

• StageTimer() — context manager，透過 time.perf_counter()(單調時鐘)測量一次呼叫的實際耗時；.build_event(schema, run_id, stage) 將其轉為消毒過的事件。
• build_stage_timing_event(...) — 較底層的建構函式；驗證 stage/status。
• summarize_stage_durations(events) — 將一串事件彙總成各階段的 count/total/avg/min/max 耗時，依總耗時由大到小排序。
• TAG_STAGE_TIMING — 對應的 TaggedOutput 標籤。

dataflow_logs.py

唯讀的 Cloud Logging 讀取器，讀取真實 Dataflow 任務自身的紀錄(resource.type="dataflow_step")——每個日誌來源(job messages、worker、harness、docker、kubelet、shuffler、autoscaling、system)都在這同一個 resource type 之下，透過 log_name 區分。

• get_dataflow_job_logs(...) — 讀取一頁。
• get_all_dataflow_job_logs(...) — 自動翻頁，上限為 max_entries(預設 5000)，避免「抓取全部」的呼叫變成無界迴圈。
• 刻意不做消毒：這是 Dataflow 原始日誌內容的完整、單純轉發——與在原生 Dataflow console 點選某個 step 時看到的內容完全相同。若需要保證消毒過的事件串流，請搭配 execution_log 使用。

最簡單的啟用方式

pipeline/parsing.py、pipeline/validation.py、pipeline/aggregation.py 已經內建 execution_log/stage_timing 的接線——只要在 build_pipeline(...) 傳入 enable_logging=True，回傳的 dict 就會多出 "execution_log"/"stage_timing" 這兩個 PCollection。省略此參數(預設行為)與此功能出現前完全相同。