The Apache Flink Runner can be used to execute Beam pipelines using Apache Flink. For execution you can choose between a cluster execution mode (e.g. Yarn/Kubernetes/Mesos) or a local embedded execution mode which is useful for testing pipelines.
The Flink Runner and Flink are suitable for large scale, continuous jobs, and provide:
It is important to understand that the Flink Runner comes in two flavors:
You may ask why there are two Runners?
Beam and its Runners originally only supported JVM-based languages (e.g. Java/Scala/Kotlin). Python and Go SDKs were added later on. The architecture of the Runners had to be changed significantly to support executing pipelines written in other languages.
If your applications only use Java, then you should currently go with the classic Runner. Eventually, the portable Runner will replace the classic Runner because it contains the generalized framework for executing Java, Python, Go, and more languages in the future.
If you want to run Python pipelines with Beam on Flink you want to use the portable Runner. For more information on portability, please visit the Portability page.
Consequently, this guide is split into parts to document the classic and the portable functionality of the Flink Runner. In addition, Python provides convenience wrappers to handle the full lifecycle of the runner, and so is further split depending on whether to manage the portability components automatically (recommended) or manually. Please use the switcher below to select the appropriate mode for the Runner:
If you want to use the local execution mode with the Flink Runner you don't have to complete any cluster setup. You can simply run your Beam pipeline. Be sure to set the Runner to FlinkRunnerPortableRunner.
To use the Flink Runner for executing on a cluster, you have to setup a Flink cluster by following the Flink Setup Quickstart.
{{< paragraph class=“language-java” >}} You must specify your dependency on the Flink Runner in your pom.xml or build.gradle. Use the Beam version and the artifact id from the above table. For example: {{< /paragraph >}}
{{< highlight java >}} org.apache.beam beam-runners-flink-1.6 {{< param release_latest >}} {{< /highlight >}}
{{< paragraph class=“language-py” >}} You will need Docker to be installed in your execution environment. To run an embedded flink cluster or use the Flink runner for Python < 3.6 you will also need to have java available in your execution environment. {{< /paragraph >}}
{{< paragraph class=“language-portable” >}} You will need Docker to be installed in your execution environment. {{< /paragraph >}}
{{< paragraph class=“language-java” >}} For executing a pipeline on a Flink cluster you need to package your program along with all dependencies in a so-called fat jar. How you do this depends on your build system but if you follow along the Beam Quickstart this is the command that you have to run: {{< /paragraph >}}
{{< highlight java >}} $ mvn package -Pflink-runner {{< /highlight >}}
{{< paragraph class=“language-java” >}} Look for the output JAR of this command in the install apache_beam``target` folder. {{< /paragraph >}}
{{< paragraph class=“language-java” >}} The Beam Quickstart Maven project is setup to use the Maven Shade plugin to create a fat jar and the -Pflink-runner argument makes sure to include the dependency on the Flink Runner. {{< /paragraph >}}
{{< paragraph class=“language-java” >}} For running the pipeline the easiest option is to use the flink command which is part of Flink: {{< /paragraph >}}
{{< paragraph class=“language-java” >}} $ bin/flink run -c org.apache.beam.examples.WordCount /path/to/your.jar --runner=FlinkRunner --other-parameters {{< /highlight >}}
{{< paragraph class=“language-java” >}} Alternatively you can also use Maven's exec command. For example, to execute the WordCount example: {{< /paragraph >}}
{{< highlight java >}} mvn exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount
-Pflink-runner
-Dexec.args=“--runner=FlinkRunner
--inputFile=/path/to/pom.xml
--output=/path/to/counts
--flinkMaster=
--filesToStage=target/word-count-beam-bundled-0.1.jar” {{< /highlight >}}
{{< paragraph class=“language-java” >}} If you have a Flink JobManager running on your local machine you can provide localhost:8081 for flinkMaster. Otherwise an embedded Flink cluster will be started for the job. {{< /paragraph >}}
{{< paragraph class=“language-py” >}} 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, one could run: {{< /paragraph >}}
{{< highlight py >}} import apache_beam as beam from apache_beam.options.pipeline_options import PipelineOptions
options = PipelineOptions([ “--runner=FlinkRunner”, “--flink_master=localhost:8081”, “--environment_type=LOOPBACK” ]) with beam.Pipeline(options) as p: ... {{< /highlight >}}
{{< paragraph class=“language-py” >}} To run on an embedded Flink cluster, simply omit the flink_master option and an embedded Flink cluster will be automatically started and shut down for the job. {{< /paragraph >}}
{{< paragraph class=“language-py” >}} The optional flink_version option may be required as well for older versions of Python. {{< /paragraph >}}
{{< paragraph class=“language-portable” >}} Starting with Beam 2.18.0, pre-built Flink Job Service Docker images are available at Docker Hub: Flink 1.8, Flink 1.9. Flink 1.10. {{< /paragraph >}}
{{< paragraph class=“language-portable” >}} 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.10_job_server:latest {{< /paragraph >}}
{{< paragraph class=“language-portable” >}} The JobService is the central instance where you submit your Beam pipeline to. The JobService will create a Flink job for the pipeline and execute the job. {{< /paragraph >}}
{{< paragraph class=“language-portable” >}} (2) Submit the Python pipeline to the above endpoint by using the PortableRunner, job_endpoint set to localhost:8099 (this is the default address of the JobService). Optionally set environment_type set to LOOPBACK. For example: {{< /paragraph >}}
{{< highlight portable >}} import apache_beam as beam from apache_beam.options.pipeline_options import PipelineOptions
options = PipelineOptions([ “--runner=PortableRunner”, “--job_endpoint=localhost:8099”, “--environment_type=LOOPBACK” ]) with beam.Pipeline(options) as p: ... {{< /highlight >}}
{{< paragraph class=“language-portable” >}} To run on a separate Flink cluster: {{< /paragraph >}}
{{< paragraph class=“language-portable” >}} (1) Start a Flink cluster which exposes the Rest interface (e.g. localhost:8081 by default). {{< /paragraph >}}
{{< paragraph class=“language-portable” >}} (2) Start JobService with Flink Rest endpoint: docker run --net=host apache/beam_flink1.10_job_server:latest --flink-master=localhost:8081. {{< /paragraph >}}
{{< paragraph class=“language-portable” >}} (3) Submit the pipeline as above. {{< /paragraph >}}
{{< highlight portable >}} import apache_beam as beam from apache_beam.options.pipeline_options import PipelineOptions
options = PipelineOptions([ “--runner=PortableRunner”, “--job_endpoint=localhost:8099”, “--environment_type=LOOPBACK” ]) with beam.Pipeline(options=options) as p: ... {{< /highlight >}}
{{< paragraph class=“language-py language-portable” >}} Note that environment_type=LOOPBACK is only intended for local testing, and will not work on remote clusters. See here for details. {{< /paragraph >}}
You can monitor a running Flink job using the Flink JobManager Dashboard or its Rest interfaces. By default, this is available at port 8081 of the JobManager node. If you have a Flink installation on your local machine that would be http://localhost:8081. Note: When you use the [local] mode an embedded Flink cluster will be started which does not make a dashboard available.
If your pipeline uses an unbounded data source or sink, the Flink Runner will automatically switch to streaming mode. You can enforce streaming mode by using the --streaming flag.
Note: The Runner will print a warning message when unbounded sources are used and checkpointing is not enabled. Many sources like PubSubIO rely on their checkpoints to be acknowledged which can only be done when checkpointing is enabled for the FlinkRunner. To enable checkpointing, please set checkpointingIntervalcheckpointing_interval to the desired checkpointing interval in milliseconds.
When executing your pipeline with the Flink Runner, you can set these pipeline options.
The following list of Flink-specific pipeline options is generated automatically from the [FlinkPipelineOptions](https://beam.apache.org/releases/javadoc/{{< param release_latest >}}/index.html?org/apache/beam/runners/flink/FlinkPipelineOptions.html) reference class:
{{< flink_java_pipeline_options >}}
{{< flink_python_pipeline_options >}}
For general Beam pipeline options see the [PipelineOptions](https://beam.apache.org/releases/javadoc/{{< param release_latest >}}/index.html?org/apache/beam/sdk/options/PipelineOptions.html) reference.
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.8.0 the minor version is 1.8.
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. The Flink community supports the last two minor versions. When support for a Flink version is dropped, it may be deprecated and removed also from Beam. To find out which version of Flink is compatible with Beam please see the table below:
For retrieving the right Flink version, see the Flink downloads page.
For more information, the Flink Documentation can be helpful.
The Beam Capability Matrix documents the capabilities of the classic Flink Runner.
The Portable Capability Matrix documents the capabilities of the portable Flink Runner.