Reusable Framework

可重用框架：beam_pipeline_toolkit

beam_pipeline_toolkit/ 是一個通用、可重用的 Apache Beam 管線框架，不含任何 Cyberbiz 專屬的商業邏輯，可以直接套用在其他「解析 → 驗證 → 彙總 → 輸出」形狀的串流管線上。

套件結構

子套件	內容
pipeline/	通用的 Beam 轉換骨架：解析、驗證、依鍵值分組彙總、輸出、Run Manifest
controlplane/	Dataflow 控制平面 API：送出 Job、停止 Job、讀取 Job 證據（狀態／計數器）
safety/	安全層：LeakGuard 禁止欄位掃描器、assert_gates_open 雲端閘門
ecommerce/	選用的電商領域擴充層，示範如何在通用框架上蓋出 Session／訂單／客戶的具體邏輯

為什麼要拆成兩層

同樣的「解析 → 驗證 → 彙總 → 輸出」骨架、以及送出／停止 Dataflow Job 的機制，幾乎每個串流管線專案都會用到。真正會變的只有商業邏輯本身：什麼樣的紀錄算合法？怎麼分組？指標列長什麼樣？這個框架把這些「會變的部分」都留成呼叫端自己傳入的函式與設定物件，框架本體不需要因為商業規則改變而跟著修改。

flowchart LR
    Y[你的商業邏輯<br/>validate_fn / key_fn / build_outcome_fn] --> T[beam_pipeline_toolkit<br/>通用骨架]
    T --> O1[你的 Pipeline]
    T --> O2[另一個完全不同形狀的 Pipeline]

Loading

最小可用範例

from beam_pipeline_toolkit.pipeline.main import build_pipeline, run_local
from beam_pipeline_toolkit.pipeline.schema import PipelineSchema

schema = PipelineSchema(
    metrics_namespace="my_app",
    allowed_stages=frozenset({"AGGREGATE"}),
    allowed_statuses=frozenset({"PASS"}),
)

def validate_fn(event: dict) -> tuple[bool, str | None]:
    if event.get("amount", -1) < 0:
        return False, "NEGATIVE_AMOUNT"
    return True, None

def key_fn(event: dict) -> tuple:
    return (event["run_id"], event["account_id"])

def build_outcome_fn(key, records):
    ...  # 組出 metrics / telemetry / invalid_summary

def pipeline_builder(raw_messages):
    return build_pipeline(
        raw_messages, schema=schema, validate_fn=validate_fn,
        key_fn=key_fn, build_outcome_fn=build_outcome_fn,
    )

run_local(input_jsonl=..., output_dir=..., pipeline_builder=pipeline_builder, ...)

更完整的雲端送出範例、ecommerce/ 擴充層的用法，請參考套件內的 beam_pipeline_toolkit/README.md。

電商擴充層（ecommerce/）

如果你的管線形狀跟電商報表很像（Session／訂單／客戶），可以直接參考或複製 ecommerce/pipeline.py，裡面示範了：

• schemas.py：SessionRecord / OrderRecord / CustomerRecord 等資料結構
• metrics.py：Decimal 精確計算的金額彙總
• order_policies.py：可替換的「訂單有效性規則」（例如是否排除已取消／已退貨訂單）
• validation.py：必填欄位驗證、Session 去重

這一層沒有寫死任何 Cyberbiz 專屬邏輯，只是讓電商形狀的管線不用從零開始。demo/ 就是把這個形狀套用到一個具體展示情境的例子——詳見 架構總覽。