Overview EN

繁體中文版

Overview

beam-pipeline-toolkit is a standalone, reusable Apache Beam pipeline framework. It contains only generic building blocks — no demo app, no dashboard, no company-specific business logic.

What it provides

Package	Responsibility
pipeline/	Generic Beam transform graph: parse → validate → aggregate → sink, plus local/cloud runner wiring
controlplane/	Dataflow control-plane API: submit a job, request cancellation, read job evidence (metadata + counters)
log/	Opt-in execution-log (sanitized exception events) + stage-timing (per-stage duration) + a read-only Cloud Logging reader
safety/	A recursive PII/secret leak guard, and a "confirm before touching the cloud" gate
ecommerce/	Optional, ready-to-adapt e-commerce domain layer built on top of the four packages above

Why the split

The same Beam transform graph (parse → validate → aggregate → sink) and the same Dataflow submit/stop/evidence mechanics show up in most event-pipeline projects. What differs is the business logic: what a valid record looks like, how records get grouped, what a metric row means. The framework pushes exactly that seam into caller-supplied functions (normalize_fn, validate_fn, key_fn, build_outcome_fn) and a couple of small config objects (PipelineSchema), so the framework itself never needs to change when business rules do.

ecommerce/ is the proof of this: it's a full, working pipeline built with the exact same public API an external caller would use — nothing in pipeline/, controlplane/, log/, or safety/ knows a session, order, or customer exists.

Data flow

raw JSON message
   │
   ▼
ParseNormalize   (pipeline/parsing.py)   -- JSON decode + normalize_fn
   │  tags: parsed | unparseable
   ▼
Validate         (pipeline/validation.py) -- caller's validate_fn
   │  tags: valid | invalid
   ▼
AggregateAndEmit (pipeline/aggregation.py) -- key_fn, GroupByKey, build_outcome_fn
   │  tags: metrics | telemetry | invalid_summary
   ▼
sinks            (pipeline/sinks.py)  -- local JSONL, or gated BigQuery/Pub/Sub

Every stage above can optionally emit stage_timing and execution_log events (see Log Module) when a PipelineSchema is passed in and enable_logging=True.

Quick start

python -m venv .venv
.venv\Scripts\activate
pip install -r requirements.txt
pytest

See beam_pipeline_toolkit/README.md for full runnable code samples (local run, cloud submission, reading job evidence).

Dependencies

Pinned in requirements.txt: apache-beam[gcp], google-auth, requests, pytest, build. Exact transitive versions are frozen in requirements.lock.txt.