Running the Apache Beam samples With Apache Flink

Check the Supported Versions on the Getting Started With Beam page to find the latest supported Flink version for your Hop version.

For example, for Hop 1.2, the latest currently supported version is 1.13. Make sure to download Flink for a recent and JDK 8 compatible Scale version. For Flink 1.3.6, this is Scala 2.11.

Download your selected Flink version and unzip to a convenient location.

Apache Flink Download

To keep things as simple as possible, we’ll run a local single node Flink cluster with a single command.

In the folder where you unzipped Flink to, run:

bin/start-cluster.sh

Your output should look similar to the one below:

  1. Starting cluster.
  2. Starting standalonesession daemon on host <HOSTNAME>.
  3. Starting taskexecutor daemon on host <HOSTNAME>.

The cluster shouldn’t take more than a couple of seconds to start. Once Flink is available, you’ll be able to access your Flink Dashboard at http://localhost:8081/

Apache Flink Dashboard

In Hop Gui’s metadata perspective for the Samples project, edit the Flink pipeline run configuration and make sure the Fat jar file location (the very last option) points to the Hop fat jar you created earlier in the prerequisites.

Apache Beam - Flink run configuration

From Hop GUI

Running Hop pipelines on embedded Flink through Hop Gui will work just fine but is intended for testing purposes and won’t show in your Flink dashboard. You can leave the default Flink master to [local] to run the embedded Flink engine from Hop Gui.

Set your Flink master to your cluster’s master. For embedded Flink, [local] will do.

Go back to the data orchestration perspective and run one of the Beam pipelines in the samples project. In this example, we used samples/beam/pipelines/generate-synthetic-data.hpl

When you start your pipeline from Hop Gui, it will appear in your Flink Dashboard.

Apache Flink Dashboard - width="90%"

In a real-world setup, you’ll run your Flink pipelines from the Flink master through flink run.

Set your Flink master to [auto] and export your Hop metadata again (see prerequisites).

Apache Beam - Flink run configuration - master

Unlike Spark you can not pass java options at runtime to the TaskManager. So we also want to set the PROJECT_HOME variable in the run configuration. This variable is used during execution to know where the source files are. (Metadata perspective → Pipeline Run Configuration → Flink → Variables)

Apache Beam - Flink run configuration - master

Use a command like the one below to pass all the information required by flink run.

  1. bin/flink run \
  2.   --class org.apache.hop.beam.run.MainBeam \
  3.   --parallelism 2 \
  4.   /opt/flink/hop-fat-jar.jar \
  5.   <PATH>/hop/config/projects/samples/beam/pipelines/generate-synthetic-data.hpl \
  6.   /opt/flink/hop-metadata.json \
  7.   Flink

With your Hop and Flink set up correctly, your output will look similar to what’s shown below:

  1. Argument 1 : Pipeline filename (.hpl)   : /home/bart/git-hop/hop/assemblies/client/target/hop/config/projects/samples/beam/pipelines/generate-synthetic-data.hpl
  2. Argument 2 : Metadata filename (.json)  : /opt/flink/hop-metadata.json
  3. Argument 3 : Pipeline run configuration : Flink
  4. >>>>>> Initializing Hop...
  5. Hop configuration file not found, not serializing: /home/bart/Projects/flink/flink-1.13.5/config/hop-config.json
  7. >>>>>> Loading pipeline metadata
  8. >>>>>> Building Apache Beam Pipeline...
  9. >>>>>> Found Beam Input transform plugin class loader
  10. >>>>>> Pipeline executing starting...
  13. 2022/02/11 12:50:25 - General - Created Apache Beam pipeline with name 'generate-synthetic-data'
  14. 2022/02/11 12:50:26 - General - Handled transform (ROW GENERATOR) : 100M rows
  15. 2022/02/11 12:50:26 - General - Handled generic transform (TRANSFORM) : random data, gets data from 1 previous transform(s), targets=0, infos=0
  16. 2022/02/11 12:50:26 - General - Handled transform (OUTPUT) : generate-synthetic-data, gets data from random data
  17. 2022/02/11 12:50:26 - General - Executing this pipeline using the Beam Pipeline Engine with run configuration 'Flink'
  18. Job has been submitted with JobID 83f34cefa8d061994b7028df2dcebfcd
  19. Program execution finished
  20. Job with JobID 83f34cefa8d061994b7028df2dcebfcd has finished.
  21. Job Runtime: 129625 ms
  22. Accumulator Results:
  23. - __metricscontainers (org.apache.beam.runners.core.metrics.MetricsContainerStepMap): {
  24.   "metrics": {
  25.     "attempted": [{
  26.       "urn": "beam:metric:user:sum_int64:v1",
  27.       "type": "beam:metrics:sum_int64:v1",
  28.       "payload": "Ag==",
  29.       "labels": {
  30.         "NAMESPACE": "startBundle",
  31.         "NAME": "random data",
  32.         "PTRANSFORM": "random data/ParMultiDo(Transform)"
  33.       }
  34.     }, {
  35.       "urn": "beam:metric:user:sum_int64:v1",
  36.       "type": "beam:metrics:sum_int64:v1",
  37.       "payload": "oI0G",
  38.       "labels": {
  39.         "NAMESPACE": "read",
  40.         "NAME": "random data",
  41.         "PTRANSFORM": "random data/ParMultiDo(Transform)"
  42.       }
  43.     }, {
  44.       "urn": "beam:metric:user:sum_int64:v1",
  45.       "type": "beam:metrics:sum_int64:v1",
  46.       "payload": "Ag==",
  47.       "labels": {
  48.         "NAMESPACE": "init",
  49.         "NAME": "random data",
  50.         "PTRANSFORM": "random data/ParMultiDo(Transform)"
  51.       }
  52.     }, {
  53.       "urn": "beam:metric:user:sum_int64:v1",
  54.       "type": "beam:metrics:sum_int64:v1",
  55.       "payload": "oI0G",
  56.       "labels": {
  57.         "NAMESPACE": "written",
  58.         "NAME": "random data",
  59.         "PTRANSFORM": "random data/ParMultiDo(Transform)"
  60.       }
  61.     }, {
  62.       "urn": "beam:metric:user:sum_int64:v1",
  63.       "type": "beam:metrics:sum_int64:v1",
  64.       "payload": "oI0G",
  65.       "labels": {
  66.         "NAMESPACE": "output",
  67.         "NAME": "generate-synthetic-data",
  68.         "PTRANSFORM": "BeamOutputTransform/generate-synthetic-data/ParMultiDo(HopToString)"
  69.       }
  70.     }, {
  71.       "urn": "beam:metric:user:sum_int64:v1",
  72.       "type": "beam:metrics:sum_int64:v1",
  73.       "payload": "oI0G",
  74.       "labels": {
  75.         "NAMESPACE": "read",
  76.         "NAME": "generate-synthetic-data",
  77.         "PTRANSFORM": "BeamOutputTransform/generate-synthetic-data/ParMultiDo(HopToString)"
  78.       }
  79.     }]
  80.   }
  81. }
  83. 2022/02/11 12:52:45 - General - Beam pipeline execution has finished.
  84. >>>>>> Execution finished...

After your pipeline finishes and the flink run command ends, your Flink dashboard will show a new entry in the ‘Completed Job List’. You can follow up any running applications in the ‘Running Job List’ and drill down into their execution details while running.

Apache Hop - Flink runner finished