Apache Beam Python YAML CSV Transform

Run a declarative Apache Beam YAML pipeline with the Python SDK on a daily schedule to download an orders CSV, filter and reshape it, and write the result back to Kestra internal storage keyed on the run date. This blueprint solves the everyday data engineering problem of turning a portable Beam pipeline into a scheduled, retried, alerted, and backfill-safe job without hand-rolling cron, glue scripts, or a brittle string-built filter. No transform code, only YAML.

Filtering a small CSV is below the threshold where Beam earns its weight, so treat this sample as a stand-in for a large partitioned source. Beam YAML pays off at volume, or when you graduate the same declarative pipeline to a portable runner such as Flink, Spark, or Dataflow without rewriting the transforms.

How it works

1. The daily_transform trigger (io.kestra.plugin.core.trigger.Schedule) fires at 05:00 with an explicit Etc/UTC timezone so the run time does not drift with the server zone or DST.
2. download_orders (io.kestra.plugin.core.http.Download) fetches inputs.source_url into internal storage as orders.csv, with retry and a timeout because a public-host download is the most likely transient failure.
3. write_pipeline (io.kestra.plugin.core.storage.Write) renders a Beam YAML pipeline that reads the CSV, keeps only orders whose total meets the threshold, selects a subset of fields, and writes high_value_orders.
4. transform_orders (io.kestra.plugin.beam.RunPipeline) runs that pipeline with sdk: PYTHON and beamRunner: DIRECT. The threshold is injected as the MIN_AMOUNT env var and read via os.environ inside the Filter callable, not concatenated into the filter code. Results are captured through outputFiles.
5. archive_partition (io.kestra.plugin.core.storage.Write) records the result under a date-partitioned key, and log_result (io.kestra.plugin.core.log.Log) prints the run date and shard URIs.

What you get

• A scheduled, declarative Beam YAML transform that needs zero transform code.
• Backfill-safe partitioning: the run date comes from {{ trigger.date ?? execution.startDate }} and the partition key is built from that same value, so replays land in the right partition.
• A typed FLOAT threshold that Kestra validates before it reaches the pipeline, passed safely as a parameter rather than concatenated into Python.
• Built-in retry and timeout on the download and pipeline tasks, plus a Slack failure alert.
• A pinned apache/beam_python3.11_sdk:2.71.0 image for reproducible runs.

Who it's for

• Data engineers running Beam pipelines who want scheduling, retries, and alerting without bespoke glue.
• Platform teams standardizing portable Beam YAML pipelines across DIRECT, Flink, Spark, and Dataflow runners.
• Analytics teams who need backfillable, partition-correct daily CSV transforms.

Why orchestrate this with Kestra

Beam's own runner executes a single pipeline, but it does not schedule, retry across runs, alert on failure, or wire one pipeline's output into the next step. Kestra adds an event-driven Schedule trigger, per-task retry with wall-clock timeout, an errors block that pages Slack on failure, and full execution lineage across download_orders, transform_orders, and the downstream archive and log tasks. The declarative YAML flow keeps the whole job version-controlled and reproducible, filling the gap Beam's scheduler cannot: backfill-aware orchestration around the pipeline.

Prerequisites

• Docker available to the Kestra worker, with the pinned Beam image.
• Outbound HTTPS so download_orders can fetch the source CSV.

Secrets

• SLACK_WEBHOOK: used only by the alert_failure errors block to post a failure notification. The sample dataset is public, so the transform itself needs no credentials. To read from a private source, store the credential as a secret and reference it with {{ secret('NAME') }}.

Quick start

1. Add the flow. It runs on its daily schedule, or trigger it manually.
2. Optionally adjust min_amount to change the filter threshold (typed FLOAT).
3. Inspect the transform_orders Outputs tab and download the high_value_orders CSV.

How to extend

• Point source_url at a date-templated path built from the run_date variable to make the schedule genuinely incremental.
• Add more Beam YAML transforms (joins, aggregations, MapToFields projections) to the pipeline body in write_pipeline.
• Graduate to a portable runner such as Flink, Spark, or Dataflow and set options.environment_type (LOOPBACK for local execution without a Docker socket mount, DOCKER for distributed sibling workers).
• Replace the Slack alert with another notification plugin, or push the filtered output to object storage or a warehouse.

Links

• Apache Beam plugin docs: https://kestra.io/plugins/plugin-beam
• Beam YAML framework: https://beam.apache.org/documentation/sdks/yaml/
• Beam YAML transforms reference: https://beam.apache.org/releases/yamldoc/current/