概要

Apache Flink Runnerは、Apache Flinkを使用してBeamパイプラインを実行するために使用できます。実行には、クラスタ実行モード（例：Yarn/Kubernetes/Mesos）またはパイプラインのテストに役立つローカル埋め込み実行モードを選択できます。

Flink RunnerとFlinkは大規模な連続ジョブに適しており、以下を提供します。

• バッチ処理とデータストリーミングプログラムの両方をサポートするストリーミングファーストのランタイム
• 非常に高いスループットと低いイベントレイテンシを同時にサポートするランタイム
• *exactly-once*処理保証による耐障害性
• ストリーミングプログラムにおける自然なバックプレッシャー
• インメモリとアウトオブコアのデータ処理アルゴリズム間の効率的かつ堅牢な切り替えのためのカスタムメモリ管理
• YARNおよびApache Hadoopエコシステムの他のコンポーネントとの統合

Apache Flink Runnerの使用

Flink Runnerには2つの種類があることを理解することが重要です。

1. Java（および他のJVMベースの言語）のみをサポートする元の*クラシックRunner*
2. Java/Python/Goをサポートする新しい*ポータブルRunner*

なぜ2つのRunnerがあるのか疑問に思うかもしれません。

BeamとそのRunnerは、もともとJVMベースの言語（例：Java/Scala/Kotlin）のみをサポートしていました。PythonとGo SDKは後で追加されました。他の言語で記述されたパイプラインの実行をサポートするために、Runnerのアーキテクチャを大幅に変更する必要がありました。

アプリケーションでJavaのみを使用する場合は、現在、クラシックRunnerを使用する必要があります。最終的には、ポータブルRunnerは、Java、Python、Go、および将来のより多くの言語を実行するための汎用フレームワークが含まれているため、クラシックRunnerに取って代わるでしょう。

FlinkでBeamを使用してPythonパイプラインを実行する場合は、ポータブルRunnerを使用する必要があります。ポータビリティの詳細については、ポータビリティのページをご覧ください。

したがって、このガイドは、Flink Runnerのクラシック機能とポータブル機能を説明するために、部分的に分割されています。さらに、Pythonはランナーのライフサイクル全体を処理するための便利なラッパーを提供するため、ポータビリティコンポーネントを自動的に管理するか（推奨）、手動で管理するかに応じてさらに分割されます。以下のスイッチャーを使用して、Runnerに適切なモードを選択してください。

前提条件とセットアップ

Flink Runnerでローカル実行モードを使用する場合、クラスタセットアップを完了する必要はありません。Beamパイプラインを実行するだけです。RunnerをFlinkRunnerPortableRunnerに設定してください。

クラスタ上で実行するためにFlink Runnerを使用するには、Flinkのセットアップクイックスタートに従ってFlinkクラスタをセットアップする必要があります。

依存関係

pom.xmlまたはbuild.gradleでFlink Runnerへの依存関係を指定する必要があります。以下の互換性テーブルからBeamバージョンとアーティファクトIDを使用してください。例えば

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-runners-flink-1.18</artifactId>
  <version>2.60.0</version>
</dependency>

実行環境にDockerがインストールされている必要があります。埋め込みflinkクラスタを実行するか、Python < 3.6 用のFlinkランナーを使用するには、実行環境でjavaも利用可能にする必要があります。

実行環境にDockerがインストールされている必要があります。

Flinkクラスタ上でのBeamパイプラインの実行

Flinkクラスタ上でパイプラインを実行するには、いわゆるfat jarにプログラムとすべての依存関係をパッケージ化する必要があります。これを行う方法はビルドシステムによって異なりますが、Beamクイックスタートに従う場合は、実行する必要があるコマンドは次のとおりです。

$ mvn package -Pflink-runner

このコマンドの出力JARは、`target`フォルダにあります。

BeamクイックスタートMavenプロジェクトは、Maven Shadeプラグインを使用してfat jarを作成するように設定されており、 `-Pflink-runner`引数はFlink Runnerへの依存関係が含まれていることを確認します。

パイプラインを実行するための最も簡単なオプションは、Flinkの一部である `flink`コマンドを使用することです。

$ bin/flink run -c org.apache.beam.examples.WordCount /path/to/your.jar –runner=FlinkRunner –other-parameters

代わりに、Mavenのexecコマンドを使用することもできます。たとえば、WordCountの例を実行するには、

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=<flink master url> \
      --filesToStage=target/word-count-beam-bundled-0.1.jar"

ローカルマシンでFlink `JobManager`が実行されている場合は、 `flinkMaster`に `localhost:8081`を指定できます。そうでない場合、ジョブのために埋め込みFlinkクラスタが起動されます。

Flinkでパイプラインを実行するには、ランナーを `FlinkRunner`に、 `flink_master`をFlinkクラスタのマスターURLに設定します。さらに、オプションで `environment_type`を `LOOPBACK`に設定します。たとえば、ローカルflinkクラスタを起動した後、次を実行できます。

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:
    ...

埋め込みFlinkクラスタで実行するには、単に `flink_master`オプションを省略すると、埋め込みFlinkクラスタが自動的に起動およびシャットダウンされます。

古いバージョンのPythonでは、オプションの `flink_version`オプションも必要になる場合があります。

Beam 2.18.0以降、事前に構築されたFlink Job Service DockerイメージはDocker Hubで入手できます：Flink 1.16。 Flink 1.17。 Flink 1.18。

埋め込みFlinkクラスタでパイプラインを実行するには

（1）JobServiceエンドポイントを起動します： `docker run --net=host apache/beam_flink1.10_job_server:latest`

JobServiceは、Beamパイプラインを送信する中央インスタンスです。 JobServiceは、パイプラインのFlinkジョブを作成し、ジョブを実行します。

（2） `PortableRunner`、 `job_endpoint`を `localhost:8099`（これはJobServiceのデフォルトアドレスです）に設定して、上記のエンドポイントにPythonパイプラインを送信します。オプションで `environment_type`を `LOOPBACK`に設定します。例えば

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:
    ...

別のFlinkクラスタで実行するには

（1）Restインターフェースを公開するFlinkクラスタを起動します（デフォルトでは `localhost:8081`）。

（2）Flink Restエンドポイントを使用してJobServiceを起動します： `docker run --net=host apache/beam_flink1.10_job_server:latest --flink-master=localhost:8081`。

（3）上記のようにパイプラインを送信します。

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:
    ...

`environment_type = LOOPBACK`はローカルテストのみを目的としており、リモートクラスタでは機能しないことに注意してください。詳細はこちらをご覧ください。

追加情報と注意事項

ジョブの監視

Flink JobManagerダッシュボードまたはそのRestインターフェースを使用して、実行中のFlinkジョブを監視できます。デフォルトでは、これはJobManagerノードのポート `8081`で利用可能です。ローカルマシンにFlinkがインストールされている場合は、 `http：// localhost：8081`になります。注： `[local]`モードを使用すると、ダッシュボードを利用できない埋め込みFlinkクラスタが起動されます。

ストリーミング実行

パイプラインで非有界データソースまたはシンクを使用する場合、Flink Runnerは自動的にストリーミングモードに切り替えます。 `--streaming`フラグを使用して、ストリーミングモードを強制できます。

注：Runnerは、非有界ソースが使用されていて、チェックポイントが無効になっている場合、警告メッセージを出力します。 `PubSubIO`などの多くのソースは、 `FlinkRunner`でチェックポイントが有効になっている場合にのみ実行できる、チェックポイントが確認されることに依存しています。チェックポイントを有効にするには、`checkpointingInterval``checkpointing_interval`をミリ秒単位の目的のチェックポイント間隔に設定してください。

Flink Runnerのパイプラインオプション

Flink Runnerでパイプラインを実行する場合、これらのパイプラインオプションを設定できます。

Flink固有のパイプラインオプションの次のリストは、FlinkPipelineOptionsリファレンスクラスから自動的に生成されます。

一般的なBeamパイプラインオプションについては、PipelineOptionsリファレンスを参照してください。

Flinkのバージョン互換性

Flinkクラスタのバージョンは、FlinkRunnerで使用されているマイナーバージョンと一致する必要があります。マイナーバージョンは、バージョン文字列の最初の2つの数字です。たとえば、1.18.0では、マイナーバージョンは1.18です。

Beamのリリース時点で、Apache Flinkの最新バージョンを追跡するようにしています。Flinkバージョンは、Flinkコミュニティでサポートされている限り、Beamでサポートされています。Flinkコミュニティは、最新の2つのマイナーバージョンをサポートしています。 Flinkバージョンのサポートが終了すると、Beamからも非推奨になり、削除される場合があります。Beamと互換性のあるFlinkのバージョンを確認するには、以下の表を参照してください。

適切なFlinkバージョンの取得については、Flinkダウンロードページを参照してください。

詳細については、Flinkドキュメントが役立ちます。

Beamの機能

Beam機能マトリクスには、従来のFlink Runnerの機能が記載されています。

ポータブル機能マトリクスには、ポータブルFlink Runnerの機能が記載されています。