Log Module EN

繁體中文版

Log Module — beam_pipeline_toolkit/log/

Generic, opt-in logging for any pipeline built on this toolkit — independent of pipeline/'s business-logic hooks and of ecommerce/.

execution_log.py

Answers "did something unexpected happen inside a specific stage, and what kind."

• exception_event(schema, run_id, stage, exc) — builds a sanitized event for a caught exception, exposing only the exception's type name, never str(exc) (which can embed the record that triggered it). Callers still re-raise afterward — the runner's own retry/failure semantics are unaffected.
• build_execution_log_event(...) — lower-level builder; validates stage against the schema and severity against {INFO, WARNING, ERROR}.
• TAG_EXECUTION_LOG — the TaggedOutput tag every DoFn registers this stream under.

stage_timing.py

Answers "how much time did each stage spend."

• StageTimer() — context manager measuring one invocation's wall-clock duration via time.perf_counter() (monotonic); .build_event(schema, run_id, stage) turns it into a sanitized event.
• build_stage_timing_event(...) — lower-level builder; validates stage/status.
• summarize_stage_durations(events) — aggregates a list of events into per-stage count/total/avg/min/max duration, sorted biggest-spender first.
• TAG_STAGE_TIMING — the matching TaggedOutput tag.

dataflow_logs.py

A read-only Cloud Logging reader for a real Dataflow job's own logs (resource.type="dataflow_step") — every log source (job messages, worker, harness, docker, kubelet, shuffler, autoscaling, system) lives under this one resource type, distinguished by log_name.

• get_dataflow_job_logs(...) — one page.
• get_all_dataflow_job_logs(...) — auto-paginates, capped at max_entries (default 5000) so a "fetch everything" call can never become an unbounded loop.
• Not sanitized (deliberately): a thin, complete pass-through of Dataflow's own log content — the same content already visible by clicking a step in the native Dataflow console. Pair with execution_log for your own guaranteed-sanitized stream.

The easy way in

pipeline/parsing.py, pipeline/validation.py, and pipeline/aggregation.py already wire execution_log/stage_timing in — pass enable_logging=True to build_pipeline(...) and the returned dict gains "execution_log"/"stage_timing" PCollections. Omitting it (the default) is identical to behavior before this feature existed.