Pipeline Module EN

繁體中文版

Pipeline Module — beam_pipeline_toolkit/pipeline/

The generic Beam transform graph. Nothing here knows about sessions, orders, or any other business concept — every hook is a caller-supplied function.

schema.py

PipelineSchema (frozen dataclass): metrics_namespace, allowed_stages, allowed_statuses. Passed everywhere a stage needs the pipeline's telemetry vocabulary; build_telemetry_event rejects any stage/status outside these closed sets so a typo fails loudly.

parsing.py — PARSE_NORMALIZE stage

ParseNormalize (PTransform) wraps ParseNormalizeDoFn: decodes bytes/text, json.loads, then runs the caller's normalize_fn. Tags: parsed (main), unparseable (a reason code only — never raw bytes). With a schema, also emits stage_timing/execution_log.

validation.py — VALIDATE stage

Validate (PTransform) wraps ValidateDoFn: calls the caller's validate_fn(event) -> (is_valid, reason_code), increments per-record-kind valid/invalid counters, tags valid (main) / invalid. Only structural/range checks belong in validate_fn — anything needing a grouped view (e.g. duplicate detection) belongs in build_outcome_fn.

aggregation.py — AGGREGATE stage

AggregateAndEmit (PTransform): KeyBy (caller's key_fn) → GroupByKey → BuildOutcome, which calls the caller's build_outcome_fn(key, records) and expects back {"metrics": ..., "telemetry": [...], "invalid_summary": ...}. Every non-None output is passed through safety.leak_guard.assert_safe before being tagged and yielded (metrics / telemetry / invalid_summary). tag_origin(event, origin) is a small helper for marking record provenance (e.g. "valid"/"excluded") so build_outcome_fn can tell records apart inside one group.

main.py — wiring

• build_pipeline(...) — the convenience wrapper for the common case: parse → validate → tag valid/invalid → key → group → build_outcome_fn, no extra filter stage. Returns a dict of named PCollections (metrics, telemetry, invalid_summary, unparseable, plus execution_log/stage_timing if enable_logging=True).
• run_local(...) — runs pipeline_builder under DirectRunner, reading a local JSONL file and writing each named output to its own local JSONL file. Touches no cloud API.
• run_cloud(...) — submits a real streaming DataflowRunner job reading from Pub/Sub; caller must have already passed its own cloud gate. Returns {"job_id", "job_name"}.
• apply_streaming_window(...) — GlobalWindows + a one-shot AfterProcessingTime trigger in ACCUMULATING mode, so GroupByKey can fire against a bounded-per-run streaming source.

A pipeline needing an extra business-rule filter stage between validation and aggregation (e.g. an order-validity policy) should compose ParseNormalize/Validate/AggregateAndEmit directly instead of calling build_pipeline — see ecommerce/pipeline.py.

sinks.py

• write_jsonl/read_jsonl — plain local JSONL I/O, always safe.
• build_cloud_pubsub_sink/build_cloud_bigquery_sink — each takes a zero-arg gate_check callable that must raise unless the cloud gate is open, called before constructing any client. BigQuery sink is append-only, CREATE_NEVER (table must already exist).

options.py

BaseToolkitPipelineOptions registers local-mode flags (--input_jsonl, --output_dir, --run_id), cloud-mode flags (--input_subscription, --telemetry_topic, --metrics_table, --run_events_table) — none with defaults — and the cloud confirmation gate flags (--enable_cloud_mode, --execute_cloud_pipeline, --confirm_cloud_execution).

metrics.py

serialize_metric_row/decimal_to_json_string — turns Decimal fields into deterministic strings for JSON output. No metric formulas live here; that's caller/domain logic (see ecommerce/metrics.py).

telemetry.py

build_telemetry_event(...) — builds one sanitized telemetry row, validates stage/status against the schema's closed sets, then runs it through assert_safe.

run_manifest.py

build_run_manifest(...) — a sanitized summary dict for one local or cloud run (run_id, mode, runner, timing, status, counts, plus any **extra).