Apache Beam Distributed Pipeline on Flink (Portable Runner)

Run a Python Apache Beam pipeline on a distributed Apache Flink cluster using the Beam portable runner, orchestrated end to end by Kestra. A JobServer bridges the serialized Python pipeline to the Flink JobManager and TaskManager, and the worker harness executes the Python code. The flow is chained off an upstream ingestion run with a Flow trigger and aggregates the data that ingestion produced (revenue per product category with a real GroupBy shuffle), so the distributed runner is justified by real work rather than a toy transform. It solves the gap that Beam and Flink themselves do not solve: event-driven submission, pre-submit reachability checks, retries, Slack alerting, and capture of output shards.

How it works

1. The io.kestra.plugin.core.trigger.Flow trigger after_ingestion fires after the upstream flow ingest_raw_events in namespace company.team finishes with state SUCCESS.
2. download_source (io.kestra.plugin.core.http.Download) pulls source_uri into Kestra internal storage as orders.csv with retry and a PT5M timeout.
3. check_jobserver (io.kestra.plugin.core.http.Request) hits {{ inputs.flink_rest_url }}/overview to confirm the cluster is reachable before submitting, so an unreachable JobServer fails fast.
4. write_pipeline (io.kestra.plugin.core.storage.Write) renders a Beam YAML pipeline that reads the CSV, sums revenue per product category with a Combine group-by (a real shuffle), and writes the result to CSV.
5. run_on_flink (io.kestra.plugin.beam.RunPipeline) submits the pipeline with sdk: PYTHON and beamRunner: FLINK, targeting the JobServer at options.job_endpoint and pointing at the Flink REST API via runnerConfig.flinkRestUrl. Output is captured through outputFiles: ["revenue_by_category*"].
6. log_result (io.kestra.plugin.core.log.Log) surfaces the Flink dashboard URL and the captured output shards.
7. On any failure, alert_failure (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) posts a Slack alert with the execution id and Flink URL.

What you get

• Event-driven Beam on Flink submissions tied to upstream ingestion success.
• Pre-submit reachability check that turns opaque submit timeouts into clear errors.
• Task-level retries and wall-clock timeouts on download, health check, and submit.
• Captured output shards (revenue_by_category*) materialized as Kestra outputs.
• Slack alerts on failure with the execution id and Flink dashboard URL.
• A real GroupBy aggregation (revenue per category) that justifies the distributed runner.

Who it's for

• Data and platform engineers running Beam on a self-managed Flink cluster.
• Teams using the Beam portable runner who need event-driven, observable submissions.
• Operators who want declarative pipelines with retries, alerting, and lineage instead of bespoke shell wrappers around the JobServer.

Why orchestrate this with Kestra

Beam's portable runner and Flink's REST API give you a distributed execution engine, not a scheduler. They have no concept of "run after upstream ingestion succeeds", no pre-submit health check, no retry-with-backoff on a flaky JobServer, no Slack alert on failure, and no native way to capture and pass output shards to a downstream flow. Kestra fills that gap: the Flow trigger reacts to upstream state, task-level retry and timeout handle transient JobServer or network issues, the errors block guarantees alerts are never silent, and the whole pipeline is declarative YAML with full execution lineage. You keep Beam and Flink for what they are good at (distributed compute) and let Kestra own scheduling, observability, and recovery.

Prerequisites

A running Flink cluster and Beam JobServer reachable from the Kestra worker. This cannot be sandboxed, hence demo: false.

• A Flink JobManager and at least one TaskManager (for example the flink:1.18 images) with the REST API exposed (default :8082).
• A Beam Flink JobServer (apache/beam_flink1.18_job_server:2.71.0) exposing the job endpoint on :8099.
• The flow defaults to environment_type: LOOPBACK, the lowest-privilege worker harness: no Docker socket, no privileged: true, no host networking. The DOCKER alternative requires mounting /var/run/docker.sock with privileged: true and host networking, which is container-escape-equivalent, so run it only on isolated, dedicated workers. EXTERNAL connects to a pre-provisioned worker pool and avoids granting any Docker access.
• Container topology: Container A (Kestra) submits, Container B (JobServer) receives at job_endpoint, Container C (Flink TaskManager) executes, Container D (Worker) runs the Python SDK harness (the Kestra task process itself under LOOPBACK).

Secrets

• SLACK_WEBHOOK: used only by the alert_failure error handler to post failure notifications. No cluster credentials are needed for this reference flow because the JobServer and Flink cluster are reached over the local network. If your cluster sits behind authentication, store the token as a secret and inject it with {{ secret('NAME') }} rather than hardcoding it.

Quick start

1. Stand up the Flink JobManager, TaskManager, and Beam JobServer, and confirm they answer on job_endpoint and flink_rest_url.
2. Add a SLACK_WEBHOOK secret for failure alerts.
3. Deploy this flow alongside the upstream ingest_raw_events flow, or run it manually with the default public source_uri.
4. Run ingest_raw_events. On success the after_ingestion trigger launches this pipeline.
5. Watch the Flink dashboard (logged by log_result) and the run_on_flink task logs.

How to extend

• Point source_uri at the real sink your ingestion writes, and bring your own ingest_raw_events flow.
• Switch environment_type to DOCKER or EXTERNAL for production worker pools (DOCKER needs a mounted Docker socket plus privileged: true and host networking, so run it only on isolated workers).
• Tune parallelism to match your TaskManager slots, and version-match the SDK and JobServer images (for example apache/beam_flink1.18_job_server:2.71.0 with apache/beam_python3.11_sdk:2.71.0).
• Replace the Combine group-by with your own Beam YAML transforms, or write results to a warehouse instead of CSV.
• Fan out to downstream flows by reading outputs.run_on_flink.outputFiles from a Subflow trigger.

Links

• Apache Beam plugin: https://kestra.io/plugins/plugin-beam
• Beam Flink runner: https://beam.apache.org/documentation/runners/flink/
• Beam portability and JobServer: https://beam.apache.org/roadmap/portability/
• Beam YAML framework: https://beam.apache.org/documentation/sdks/yaml/