Datastream development packaging issues

Flink のコア依存関係とアプリケーション固有の依存関係

すべての Flink アプリケーションは、一連の関連ライブラリに依存しており、少なくとも Flink の API を含む必要があります。多くのアプリケーションは、コネクタ関連のライブラリ（Kafka、Cassandra など）にも依存しています。Flink アプリケーションの実行時、分散環境での実行かローカル IDE でのテストかにかかわらず、Flink のランタイム関連の依存関係が必要です。

ユーザー定義アプリケーションを実行するほとんどのシステムと同様に、Flink には大きく分けて 2 つのカテゴリの依存関係があります。

Flink コア依存関係：Flink 自体は、システムの実行に必要な一連のクラスと依存関係で構成されています。これには、コーディネーター、ネットワーク、チェックポイント、フォールトトレランス、API、オペレーター（ウィンドウなど）、リソース管理などが含まれます。これらのすべてのクラスと依存関係の集合が Flink ランタイムのコアを形成し、Flink アプリケーションの起動時に存在している必要があります。これらのコアクラスと依存関係は flink-dist jar にパッケージ化されています。これらは Flink の lib フォルダの一部であり、Flink ベースコンテナイメージの一部です。これらの依存関係は、Flink にとって、Java の実行に必要な String や List などのクラスを含むコアライブラリ（rt.jar、charset.jar など）が Java にとってそうであるのと同じ存在です。Flink のコア依存関係には、コネクタや拡張ライブラリ（CEP、SQL、ML など）は含まれていません。これにより、Flink のコア依存関係を可能な限り小さく保ち、デフォルトでのクラスパス内の過剰な依存関係を回避し、依存関係の競合を低減しています。

ユーザーアプリケーション依存関係：特定のユーザーアプリケーションが必要とするすべてのコネクタ、フォーマット、または拡張ライブラリを指します。ユーザーアプリケーションは通常 jar ファイルとしてパッケージ化され、アプリケーションコードとともに必要なコネクタおよびライブラリの依存関係が含まれます。ユーザーアプリケーション依存関係には、Flink DataStream API およびランタイム依存関係を含めないでください。これらは既に Flink のコア依存関係に含まれています。

依存関係の設定手順

1. 基本依存関係の追加

各 Flink アプリケーションの開発には、少なくとも関連する API への基本依存関係を追加する必要があります。

プロジェクトを手動で設定する場合、Java/Scala API への依存関係を追加する必要があります（ここでは Maven を使用した例を示しますが、他のビルドツール（Gradle、SBT など）でも同じ依存関係を使用できます）。

重要事項：これらの依存関係はすべてスコープが「provided」に設定されている点に注意してください。これは、コンパイル時には必要ですが、プロジェクトが生成するアプリケーション JAR ファイルにパッケージ化すべきではないことを意味します。これらの依存関係は Flink コア依存関係であり、実際の実行時には既にロードされています。

依存関係を「provided」スコープに設定することを強く推奨します。「provided」に設定しないと、最良の場合はすべての Flink コア依存関係も含まれるため JAR ファイルが肥大化します。最悪の場合は、アプリケーション JAR ファイルに追加された Flink コア依存関係が独自の依存関係の一部とバージョン競合を起こす可能性があります（通常は Flink のリバースクラスローディングメカニズムにより回避されます）。

IntelliJ IDEA に関する注意：IntelliJ IDEA でアプリケーションを実行するには、実行構成で「Include dependencies with "Provided" scope」オプションのチェックボックスを有効にする必要があります。このオプションが表示されない場合（古いバージョンの IntelliJ IDEA を使用している可能性があります）、アプリケーションの main() メソッドを呼び出すテストケースを作成するという簡単な回避策があります。

2. コネクタおよびライブラリ依存関係の追加

ほとんどのアプリケーションは、Kafka や Cassandra などの特定のコネクタまたはライブラリを実行時に必要とします。これらのコネクタは Flink のコア依存関係には含まれておらず、追加の依存関係としてアプリケーションに追加する必要があります。

アプリケーションコードとすべての依存関係を jar-with-dependencies 形式のアプリケーション JAR としてパッケージ化することを推奨します。アプリケーション JAR ファイルは既存の Flink クラスターに送信するか、Flink アプリケーションのコンテナイメージに追加できます。

Maven ジョブテンプレートから作成されたプロジェクト（後述の Maven ジョブテンプレートセクションを参照）では、mvn clean package コマンドにより依存関係が自動的にアプリケーション JAR ファイルに追加されます。テンプレートを使用せずに設定する場合は、Maven Shade Plugin（設定例は付録を参照）を使用して依存関係を含む JAR ファイルを構築することを推奨します。

重要事項：Maven（および他のビルドツール）が依存関係をアプリケーション JAR に正しくパッケージ化するには、アプリケーション依存関係のスコープを「compile」に設定する必要があります（コア依存関係とは異なり、コア依存関係のスコープは「provided」に指定する必要があります）。

注意事項
Scala バージョン

異なるバージョンの Scala（2.11、2.12 など）には互換性がありません。したがって、Scala 2.11 対応の Flink を Scala 2.12 を使用するアプリケーションに使用することはできません。

Scala に依存（または推移的に依存）するすべての Flink 依存関係には、ビルドされた Scala バージョンのサフィックスが付いています（例：flink-streaming-scala_2.11）。

Java のみで開発する場合は任意の Scala バージョンを選択でき、Scala で開発する場合はアプリケーションの Scala バージョンに一致する Flink 依存関係のバージョンを選択する必要があります。

注意：Scala 2.12.8 以降は以前の 2.12.x バージョンと互換性がないため、Flink プロジェクトは 2.12.x バージョンを 2.12.8 以降にアップグレードできません。ユーザーはローカル環境で対応する Scala バージョンの Flink をコンパイルできます。これを行うには、ビルド時にバイナリ互換性チェックをスキップするため -Djapicmp.skip を追加する必要があります。

Hadoop 依存関係

一般ルール：Hadoop 関連の依存関係をアプリケーションに直接追加しないでください。（唯一の例外は、Flink の Hadoop 互換パッケージで既存の Hadoop 入出力フォーマットを使用する場合です。）

Flink を Hadoop と連携して使用する場合、Hadoop をアプリケーション依存関係として追加するのではなく、Hadoop が依存する Flink の起動項目を含める必要があります。Flink は HADOOP_CLASSPATH 環境変数で指定された Hadoop 依存関係を使用します。この変数は以下のように設定できます。

export HADOOP_CLASSPATH=hadoop classpath

この設計には主に 2 つの理由があります。

Hadoop とのやり取りの一部は、Flink のコアモジュール内で、ユーザーアプリケーションの起動前に行われる場合があります。たとえば、チェックポイント用の HDFS の設定、Hadoop の Kerberos トークンによる認証、YARN へのデプロイなどです。
Flink のリバースクラスローディングメカニズムは、多くの推移的依存関係をコア依存関係から隠蔽します。これは Flink 自体のコア依存関係だけでなく、Hadoop の依存関係にも適用されます。これにより、アプリケーションは同じ依存関係の異なるバージョンを依存関係の競合なしに使用できます（信頼してください。Hadoop の依存関係ツリーは非常に巨大なため、これは重要なことです。）
IDE 内のテストや開発中に Hadoop 依存関係（HDFS アクセスなど）が必要な場合は、これらの依存関係のスコープを test または provided に設定してください。

特定識別子を持つテーブルコネクタ/フォーマットファクトリを読み込むメカニズム。各テーブルコネクタ/フォーマットの org.apache.flink.table.factories.Factory という名前の SPI リソースファイルは同じディレクトリ（META-INF/services 配下）に配置されているため、複数のテーブルコネクタ/フォーマットを使用する uber-jar を構築する際、これらのリソースファイルが互いに上書きし合い、Flink がファクトリークラスを正しく読み込めなくなります。

この場合、Maven Shade Plugin の ServicesResourceTransformer により META-INF/services ディレクトリ配下のリソースファイルを変換する方法を推奨します。以下に、flink-sql-connector-hive-3.1.2 コネクタと flink-parquet フォーマットを含む pom.xml ファイルの設定例を示します。

ServicesResourceTransformer を設定後、プロジェクトが uber-jar を構築する際、META-INF/services ディレクトリ内のこれらのリソースファイルは上書きされる代わりに統合されます。

Maven ジョブテンプレート
多くの繰り返しの設定作業を削減できるため、この方法の使用を強く推奨します。

事前要件
環境要件は、Maven 3.0.4（またはそれ以上）および Java 8.x がインストールされていることのみです。

プロジェクトの作成
以下の 2 つの方法のいずれかを使用してプロジェクトを作成します。

この方法では、新しく作成するプロジェクトに名前を付けることができます。groupId、artifactId、およびパッケージ名が対話的に尋ねられます。

クイックスタートスクリプトの実行

$ curl https://flink.apache.org/q/quickstart.sh | bash -s 1.12.3
このプロジェクトを IDE にインポートして開発およびテストを行うことを推奨します。IntelliJ IDEA は Maven プロジェクトをネイティブにサポートしています。Eclipse を使用している場合は、m2e プラグインを使用して Maven プロジェクトをインポートできます。Eclipse の一部のバンドルにはデフォルトでこのプラグインが含まれていますが、含まれていない場合は手動でインストールする必要があります。

注意：Flink にはデフォルトの Java JVM ヒープサイズが小さすぎる場合があります。手動で増やす必要があります。Eclipse では、[Run Configurations] -> [Arguments] を選択し、[VM Arguments] ボックスに -Xmx800m と入力してください。IntelliJ IDEA では、[Help] | [Edit Custom VM Options] オプションメニューを使用して JVM オプションを変更する方法を推奨します。詳細は関連記事を参照してください。

プロジェクトのビルド
プロジェクトをビルドまたはパッケージ化するには、プロジェクトディレクトリに移動して mvn clean package コマンドを実行します。実行の結果、アプリケーションを含む JAR ファイル target/-.jar が生成され、依存関係として追加されたコネクタおよびライブラリも含まれます。

注意：StreamingJob 以外のクラスをアプリケーションのメインクラスまたはエントリポイントとして使用している場合は、pom.xml ファイルの mainClass 設定を適切に変更することを推奨します。これにより、Flink は追加でメインクラスを指定することなく、JAR ファイルから直接アプリケーションを実行できます。

付録：依存関係を含む JAR ファイル構築用テンプレート
コネクタおよびライブラリに必要なすべての依存関係を含むアプリケーション JAR を構築するには、以下の shade プラグインの定義を使用できます。