////
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at
  http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
////
[[BeamFlinkPipelineEngine]]
:imagesdir: ../assets/images
:openvar: ${
:closevar: }
:description: Apache Hop supports running pipelines on Apache Flink using an Apache Beam Flink runner. This page describes how to configure this runner.
= Apache Beam Flink Pipeline Engine

== Beam Flink

This runner allows you to run Hop pipelines on https://flink.apache.org[Apache Flink] version 1.13.

The Flink runner supports two modes: Local Direct Flink Runner and Flink Runner.

The Flink Runner and Flink are suitable for large scale, continuous jobs, and provide:

* A streaming-first runtime that supports both batch processing and data streaming programs
* A runtime that supports very high throughput and low event latency at the same time
* Fault-tolerance with exactly-once processing guarantees
* Natural back-pressure in streaming programs
* Custom memory management for efficient and robust switching between in-memory and out-of-core data processing algorithms
* Integration with YARN and other components of the Apache Hadoop ecosystem

Check the https://beam.apache.org/documentation/runners/flink/[Apache Beam Flink runner docs] for more information.

=== Options

[options="header",cols="1,3,1"]
|===
|Option|Description|Default
|The Flink master|Address of the Flink Master where the Pipeline should be executed.
Can either be of the form "host:port" or one of the special values [local], [collection] or [auto]|auto
|Parallelism|The pipeline wide maximum degree of parallelism to be used.
The maximum parallelism specifies the upper limit for dynamic scaling and the number of key groups used for partitioned state.|-1
|Checkpointing interval|The interval in milliseconds at which to trigger checkpoints of the running pipeline.|No checkpointing, -1
|Checkpointing timeout (ms)|The maximum time in milliseconds that a checkpoint may take before being discarded.|-1
|Minimum pause between checkpoints|The minimal pause in milliseconds before the next checkpoint is triggered|-1
|Fail on checkpointing errors| Sets the expected behaviour for tasks in case that they encounter an error in their checkpointing procedure.
If this is set to true, the task will fail on checkpointing error.
If this is set to false, the task will only decline a the checkpoint and continue running|true
|Number of execution retries|Sets the number of times that failed tasks are re-executed.
A value of zero effectively disables fault tolerance.
A value of -1 indicates that the system default value (as defined in the configuration) should be used.|-1
|Execution retry delay (ms)|Sets the delay in milliseconds between executions.
A value of {@code -1} indicates that the default value should be used.|-1
|Object re-use|Sets the behavior of reusing objects.
Enabling the object reuse mode will instruct the runtime to reuse user objects for better performance.|false
|Disable metrics|Disable Beam metrics in Flink Runner|-1
|Retain externalized checkpoints on cancellation|Sets the behavior of externalized checkpoints on cancellation.|false
|Maximum bundle size|The maximum number of elements in a bundle.|1000
|Maximum bundle time (ms)|The maximum time to wait before finalising a bundle (in milliseconds).|1000
|Shutdown sources on final watermark|Shuts down sources which have been idle for the configured time of milliseconds.
Once a source has been shut down, checkpointing is not possible anymore.
Shutting down the sources eventually leads to pipeline shutdown (=Flink job finishes) once all input has been processed.
Unless explicitly set, this will default to Long.MAX_VALUE when checkpointing is enabled and to 0 when checkpointing is disabled.
See https://issues.apache.org/jira/browse/FLINK-2491[FLINK-2491] for progress on this issue.|
|Latency tracking interval|    Interval in milliseconds for sending latency tracking marks from the sources to the sinks.
Interval value <= 0 disables the feature.|0
|Auto watermark interval|The interval in milliseconds for automatic watermark emission.|
|Batch execution mode|Flink mode for data exchange of batch pipelines.
Reference {@link org.apache.flink.api.common.ExecutionMode}.
Set this to BATCH_FORCED if pipelines get blocked, see https://issues.apache.org/jira/browse/FLINK-10672[FLINK-10672]|P
|User agent|A user agent string as per https://tools.ietf.org/html/rfc2616[RFC2616], describing the pipeline to external services.|
|Temp location|Path for temporary files.|
|Plugins to stage (, delimited)|Comma separated list of plugins.|
|Transform plugin classes|List of transform plugin classes.|
|XP plugin classes|List of extensions point plugins.|
|Streaming Hop transforms flush interval (ms)|The amount of time after which the internal buffer is sent completely over the network and emptied.|
|Hop streaming transforms buffer size|The internal buffer size to use.|
|Fat jar file location|Fat jar location.|
|===

== Running with Flink Run

You can also execute using the 'bin/flink run' command.
There is a main class you can use with the `--class` option of the run command:

[source]
----
org.apache.hop.beam.run.MainBeam
----

It accepts 3 or 4 arguments:

|===
|Argument|Description

|1
|The filename of the pipeline to execute.

|2
|The filename of the metadata to load (JSON).
You can export metadata in the Hop GUI under the tools menu (part of the Beam plugin in ```plugins/engines/beam``` )

|3
|The name of the pipeline run configuration to use

|4 (optional)
|The name of the environment file (JSON) containing the variables you want to set in the pipeline.  The format is identical to what is used in hop-config.json or an environment configuration file.
|===

The Flink run command also needs a fat jar as an argument.
This can be generated in the Hop GUI under the tools menu or using command:

[source,bash]
----
sh hop-config.sh -fj /path/to/fat.jar
----

**Important** : project configurations, environments and these things are not valid in the context of the Flink runtime.
This is a TODO for the Hop community to think how we can do this best.
Your input is welcome.
In the meantime pass variables to the JVM by setting these in the conf/flink-conf.yml file by adding a line:

[source,yaml]
----
env.java.opts: -DPROJECT_HOME=/path/to/project-home
----

In general, it is better not to use relative paths like `{openvar}Internal.Entry.Current.Folder{closevar}` when specifying filenames when executing pipelines remotely.
It's usually better to pick a few root folders as variables.
PROJECT_HOME is as good as any variable to use.

An example Flink run command might look like this:

[source,bash]
----
bin/flink run \
  --class org.apache.hop.beam.run.MainBeam \
  --parallelism 2 \
  -D PROJECT_HOME=/my/project/home \
  /path/to/apache-hop-fat.jar \
  /my/project/home/pipeline.hpl \
  metadata-export.json \
  FlinkRunConfig
----

== Flink embedded

You can specify a master of `[local]` to run using an embedded Flink engine.
It's primarily used for testing locally.
