diff --git a/contributor-docs/README.md b/contributor-docs/README.md index 1087a74f05c7..7e4381893c93 100644 --- a/contributor-docs/README.md +++ b/contributor-docs/README.md @@ -22,6 +22,7 @@ This directory contains documentation for contributors to the Apache Beam projec - [Committer Guide](committer-guide.md): Guidelines for Beam committers regarding code review, pull request objectives, merging processes, and post-merge tasks. - [Committer Onboarding](committer-onboarding.md): A checklist for new Beam committers to set up their accounts and permissions. - [Java Dependency Upgrades](java-dependency-upgrades.md): Instructions for upgrading Java dependencies in Beam, including running linkage checkers and verification tests. +- [Local Flink Python Validation](local-flink-python.md): Instructions for running Python pipelines on a local Flink standalone cluster. - [Python Tips](python-tips.md): Tips and instructions for developing the Python SDK, including environment setup, running tests, and handling dependencies. - [RC Testing Guide](rc-testing-guide.md): A guide for testing Beam Release Candidates (RCs) against downstream projects for Python, Java, and Go SDKs. - [Release Guide](release-guide.md): A comprehensive guide for the Release Manager on how to perform a Beam release, from preparation to promotion. diff --git a/contributor-docs/local-flink-python.md b/contributor-docs/local-flink-python.md new file mode 100644 index 000000000000..6c7dfb3e2697 --- /dev/null +++ b/contributor-docs/local-flink-python.md @@ -0,0 +1,204 @@ + + +# Running Python pipelines on a local Flink cluster + +This guide describes a contributor workflow for validating Python Beam pipelines +against a real local Flink standalone cluster. It is useful when embedded Flink +is not enough, for example when validating streaming source behavior, checkpoint +boundaries, or runner-visible job state in the Flink dashboard. + +The commands assume a Unix shell (Linux, macOS, or WSL2 on Windows) with `curl`, +`tar`, and `java` on the `PATH`. + +* [What this setup validates](#what-this-setup-validates) +* [Prerequisites](#prerequisites) +* [Start a local Flink cluster](#start-a-local-flink-cluster) +* [Run a Beam Python pipeline](#run-a-beam-python-pipeline) +* [Troubleshooting](#troubleshooting) +* [Stop the cluster](#stop-the-cluster) + +## What this setup validates + +This setup runs three separate processes: + +1. A Flink standalone cluster, consisting of a JobManager and a TaskManager. +1. A Beam Flink Job Server, started by the Python `FlinkRunner`. +1. A Python SDK harness, using `--environment_type=LOOPBACK` for local + development. + +The Flink dashboard at `http://localhost:8081` shows the submitted Beam jobs. +This is different from embedded Flink mode, where the cluster is started only +for the lifetime of one job and is not useful for manual dashboard inspection. + +## Prerequisites + +Install or prepare the following: + +* Docker Desktop (optional), only for the alternative method of obtaining the + Flink distribution. +* A Unix shell: Linux, macOS, or WSL2 on Windows. +* Java 11 on the `PATH`. +* A Python environment with the Beam SDK dependencies installed. +* A Beam source checkout for the Python code under test. +* A Flink 1.20 Job Server jar built from the same Beam checkout when validating + unreleased Beam changes. + +For a source-built Job Server jar, run this command from the Beam checkout: + +```sh +./gradlew :runners:flink:1.20:job-server:shadowJar +``` + +The jar is written under: + +```text +runners/flink/1.20/job-server/build/libs/ +``` + +## Start a local Flink cluster + +Use a Flink distribution whose minor version matches a Flink version supported +by your Beam version. See the [Flink Version Compatibility](https://beam.apache.org/documentation/runners/flink/#flink-version-compatibility) +table in the Flink Runner documentation, and confirm the exact patch version on +the [Flink downloads page](https://flink.apache.org/downloads.html). This guide +uses Flink 1.20. + +Download and unpack the binary distribution: + +```sh +FLINK_VERSION=1.20.1 +curl -fLO "https://archive.apache.org/dist/flink/flink-${FLINK_VERSION}/flink-${FLINK_VERSION}-bin-scala_2.12.tgz" +tar -xzf "flink-${FLINK_VERSION}-bin-scala_2.12.tgz" -C "$HOME" +export FLINK_HOME="$HOME/flink-${FLINK_VERSION}" +``` + +Ensure these settings exist in `$FLINK_HOME/conf/config.yaml`: + +```yaml +jobmanager.rpc.address: localhost +rest.address: localhost +taskmanager.numberOfTaskSlots: 2 +``` + +Start the cluster. The JobManager and TaskManager run as background daemons: + +```sh +"$FLINK_HOME/bin/start-cluster.sh" +``` + +Verify that the JobManager and TaskManager are available: + +```sh +curl -fsS http://localhost:8081/overview +``` + +Expected output includes one TaskManager and two slots: + +```json +{"taskmanagers":1,"slots-total":2,"slots-available":2,"jobs-running":0} +``` + +You can also open the Flink dashboard in a browser: + +```text +http://localhost:8081 +``` + +### Alternative: extract Flink from the Docker image + +If a direct download is not available, copy the distribution out of the Flink +Docker image with `docker cp`: + +```sh +docker create --name flink-dist flink:1.20 +docker cp flink-dist:/opt/flink "$HOME/flink-1.20" +docker rm flink-dist +export FLINK_HOME="$HOME/flink-1.20" +``` + +A distribution copied out of a Docker image can contain the container hostname in +`conf/config.yaml`; see [Troubleshooting](#troubleshooting). + +## Run a Beam Python pipeline + +For local Python development, use `FlinkRunner`, point it at the standalone +cluster, and use `LOOPBACK` so the Python SDK harness runs in the local process. + +Use a source checkout on `PYTHONPATH` when validating unreleased Python changes. +Set paths for your environment: + +```sh +export BEAM_CHECKOUT="$HOME/beam" +export PYTHON="$HOME/beamenv/bin/python" +export FLINK_JOB_SERVER_JAR="$(find "$BEAM_CHECKOUT/runners/flink/1.20/job-server/build/libs" \ + -name 'beam-runners-flink-1.20-job-server-*.jar' | head -n 1)" +``` + +Run a small pipeline: + +```sh +printf 'to be or not to be\nbeam runs on flink\n' > /tmp/beam-flink-input.txt + +PYTHONPATH="$BEAM_CHECKOUT/sdks/python" "$PYTHON" -m apache_beam.examples.wordcount \ + --runner=FlinkRunner \ + --flink_master=localhost:8081 \ + --flink_version=1.20 \ + --flink_job_server_jar="$FLINK_JOB_SERVER_JAR" \ + --environment_type=LOOPBACK \ + --input=/tmp/beam-flink-input.txt \ + --output=/tmp/beam-flink-counts +``` + +For released Beam, omit `--flink_job_server_jar` and the `PYTHONPATH` prefix; the +`FlinkRunner` downloads a Job Server matching `--flink_version` automatically. The +source checkout and built jar are only needed to test unreleased changes. + +Check the dashboard or REST API after the run: + +```sh +curl -fsS http://localhost:8081/jobs/overview +``` + +The job should be `FINISHED`. + +## Troubleshooting + +If the TaskManager does not register, check `$FLINK_HOME/conf/config.yaml`. +When a distribution is copied out of a Docker image, the file might contain the +container hostname. Replace it with: + +```yaml +jobmanager.rpc.address: localhost +``` + +If a Python job fails on native Windows with an invalid path containing `:`, +run the Python driver and Job Server from WSL2. Some staged artifact names used +by the portable runner are valid on Linux but invalid as native Windows file +names. + +On WSL2, keep at least one shell open in the distribution while the cluster runs. +Closing the last shell can stop the distribution and its background daemons. + +If the job starts but the Python transforms do not execute, check the +environment type. `LOOPBACK` is intended for local development. For a remote +or multi-machine Flink cluster, use a containerized environment instead. + +## Stop the cluster + +Stop the local cluster when you finish collecting results: + +```sh +"$FLINK_HOME/bin/stop-cluster.sh" +``` diff --git a/website/www/site/content/en/documentation/runners/flink.md b/website/www/site/content/en/documentation/runners/flink.md index e924ccdb7bd6..cbfa42f54cef 100644 --- a/website/www/site/content/en/documentation/runners/flink.md +++ b/website/www/site/content/en/documentation/runners/flink.md @@ -93,7 +93,7 @@ from the [compatibility table](#flink-version-compatibility) below. For example: {{< highlight java >}} org.apache.beam - beam-runners-flink-1.18 + beam-runners-flink-1.20 {{< param release_latest >}} {{< /highlight >}} @@ -166,7 +166,7 @@ If you have a Flink `JobManager` running on your local machine you can provide ` To run a pipeline on Flink, set the runner to `FlinkRunner` and `flink_master` to the master URL of a Flink cluster. In addition, optionally set `environment_type` set to `LOOPBACK`. For example, -after starting up a [local flink cluster](https://ci.apache.org/projects/flink/flink-docs-release-1.18/getting-started/tutorials/local_setup.html), +after starting up a [local flink cluster](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/try-flink/local_installation/), one could run: {{< /paragraph >}} @@ -196,9 +196,8 @@ The optional `flink_version` option may be required as well for older versions o {{< paragraph class="language-portable" >}} Starting with Beam 2.18.0, pre-built Flink Job Service Docker images are available at Docker Hub: -[Flink 1.16](https://hub.docker.com/r/apache/beam_flink1.16_job_server). -[Flink 1.17](https://hub.docker.com/r/apache/beam_flink1.17_job_server). -[Flink 1.18](https://hub.docker.com/r/apache/beam_flink1.18_job_server). +[Flink 1.19](https://hub.docker.com/r/apache/beam_flink1.19_job_server). +[Flink 1.20](https://hub.docker.com/r/apache/beam_flink1.20_job_server). {{< /paragraph >}} @@ -207,7 +206,7 @@ To run a pipeline on an embedded Flink cluster: {{< /paragraph >}} {{< paragraph class="language-portable" >}} -(1) Start the JobService endpoint: `docker run --net=host apache/beam_flink1.18_job_server:latest` +(1) Start the JobService endpoint: `docker run --net=host apache/beam_flink1.20_job_server:latest` {{< /paragraph >}} {{< paragraph class="language-portable" >}} @@ -217,7 +216,7 @@ You might encounter an error message like `Caused by: java.io.IOException: Insuf This can be resolved by providing a Flink configuration file to override the default settings. You can find an example configuration file [here](https://github.com/apache/beam/blob/master/runners/flink/src/test/resources/flink-conf.yaml). To start the Job Service endpoint with your custom configuration, mount a local directory containing your Flink configuration to the `/flink-conf` path in the Docker container and pass this as `--flink-conf-dir`: -`docker run --net=host -v :/flink-conf beam-flink-runner apache/beam_flink1.18_job_server:latest --flink-conf-dir /flink-conf` +`docker run --net=host -v :/flink-conf beam-flink-runner apache/beam_flink1.20_job_server:latest --flink-conf-dir /flink-conf` {{< /paragraph >}} {{< paragraph class="language-portable" >}} @@ -240,7 +239,7 @@ with beam.Pipeline(options) as p: {{< paragraph class="language-portable" >}} -To run on a separate [Flink cluster](https://ci.apache.org/projects/flink/flink-docs-release-1.18/getting-started/tutorials/local_setup.html): +To run on a separate [Flink cluster](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/try-flink/local_installation/): {{< /paragraph >}} {{< paragraph class="language-portable" >}} @@ -248,7 +247,7 @@ To run on a separate [Flink cluster](https://ci.apache.org/projects/flink/flink- {{< /paragraph >}} {{< paragraph class="language-portable" >}} -(2) Start JobService with Flink Rest endpoint: `docker run --net=host apache/beam_flink1.18_job_server:latest --flink-master=localhost:8081`. +(2) Start JobService with Flink Rest endpoint: `docker run --net=host apache/beam_flink1.20_job_server:latest --flink-master=localhost:8081`. {{< /paragraph >}} {{< paragraph class="language-portable" >}} @@ -316,8 +315,8 @@ reference. ## Flink Version Compatibility The Flink cluster version has to match the minor version used by the FlinkRunner. -The minor version is the first two numbers in the version string, e.g. in `1.18.0` the -minor version is `1.18`. +The minor version is the first two numbers in the version string, e.g. in `1.20.0` the +minor version is `1.20`. We try to track the latest version of Apache Flink at the time of the Beam release. A Flink version is supported by Beam for the time it is supported by the Flink community.