Flink データストリーム API プログラミング ガイド #

これらの各ステップの概要を説明します。詳細はそれぞれのセクションを参照してください。Java DataStream APIの全てのコアクラスは org.apache.flink.streaming.api にあることに注意してください。

StreamExecutionEnvironmentは全てのFlinkプログラムの基礎です。StreamExecutionEnvironmentの静的メソッドを使って取得できます:

getExecutionEnvironment();

createLocalEnvironment();

createRemoteEnvironment(String host, int port, String... jarFiles);

通常、getExecutionEnvironment()を使うだけで済みます。これは、コンテキストに応じて適切な動作を行うためです: プログラムをIDE内で実行している場合や通常のJavaプログラムとして実行している場合は、ローカルマシン上でプログラムを実行するローカル環境が作成されます。プログラムからJARファイルを作成し、コマンドラインを通じて呼び出すと、Flinkクラスタマネージャーはmainメソッドを実行し、getExecutionEnvironment()はクラスタ上でプログラムを実行するための実行環境を返します。

データソースを指定するために、実行環境には様々な方法でファイルから読み取る方法がいくつかあります: データを1行ずつ読み取ることも、CSVVファイルとして読み取ることも、その他の提供されたソースを使うこともできます。テキストファイルから単に連続する行として読み取るには、次のものが使えます:

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<String> text = env.readTextFile("file:///path/to/file");

これにより、変換を適用して新しい派生DataStreamを作成できるDataStreamが得られます。

変換関数を使ってDataStream上でメソッドを呼び出すことで、変換を適用できます。例えば、map変換はこのようになります:

DataStream<String> input = ...;

DataStream<Integer> parsed = input.map(new MapFunction<String, Integer>() {
    @Override
    public Integer map(String value) {
        return Integer.parseInt(value);
    }
});

これは、元のコレクション内の全ての文字列が整数に変換されて、新しいDataStreamを作成します。

最終結果を含むDataStreamを取得したら、シンクを作成して外部システムにそれを書き込むことができます。以下はシンクを作成するためのメソッドのほんの一部の例です:

writeAsText(String path);

print();

これらの各ステップの概要を説明します。詳細はそれぞれのセクションを参照してください。Scala DataStream APIの全てのコアクラスは、 org.apache.flink.streaming.api.scala にあります。

StreamExecutionEnvironmentは全てのFlinkプログラムの基礎です。StreamExecutionEnvironmentの静的メソッドを使って取得できます:

getExecutionEnvironment()

createLocalEnvironment()

createRemoteEnvironment(host: String, port: Int, jarFiles: String*)

通常、getExecutionEnvironment()を使うだけで済みます。これは、コンテキストに応じて適切な動作を行うためです: プログラムをIDE内で実行している場合や通常のJavaプログラムとして実行している場合は、ローカルマシン上でプログラムを実行するローカル環境が作成されます。プログラムからJARファイルを作成し、コマンドラインを通じて呼び出すと、Flinkクラスタマネージャーはmainメソッドを実行し、getExecutionEnvironment()はクラスタ上でプログラムを実行するための実行環境を返します。

データソースを指定するために、実行環境には様々な方法でファイルから読み取る方法がいくつかあります: データを1行ずつ読み取ることも、CSVVファイルとして読み取ることも、その他の提供されたソースを使うこともできます。テキストファイルから単に連続する行として読み取るには、次のものが使えます:

val env = StreamExecutionEnvironment.getExecutionEnvironment()

val text: DataStream[String] = env.readTextFile("file:///path/to/file")

変換を適用して新しい派生DataStreamを作成できるDataStreamが得られます。

変換関数を使ってDataStream上でメソッドを呼び出すことで、変換を適用できます。例えば、map変換はこのようになります:

val input: DataSet[String] = ...

val mapped = input.map { x => x.toInt }

これは、元のコレクション内の全ての文字列が整数に変換されて、新しいDataStreamを作成します。

最終結果を含むDataStreamを取得したら、シンクを作成して外部システムにそれを書き込むことができます。以下はシンクを作成するためのメソッドのほんの一部の例です:

writeAsText(path: String)

print()

ソースはプログラムが入力を取り込むところです。StreamExecutionEnvironment.addSource(sourceFunction)を使ってソースをプログラムに付け加えることができます。Flinkは多数の事前実装されたソース関数が同梱されますが、非並列ソースについてはSourceFunctionを実装することで独自のソースを常に書くことができ、並列ソースについてはParallelSourceFunctionインタフェースを実装するかRichParallelSourceFunctionを拡張することで独自のソースを常に書くことができます。

StreamExecutionEnvironmentからアクセス可能な幾つかの事前定義されたストリームソースがあります:

ファイルベース:

• readTextFile(path) - テキストファイル、つまりTextInputFormatの仕様を考慮するファイルを行ごとに読み込み、それらを文字列として返します。

• readFile(fileInputFormat, path) - 指定されたファイル入力フォーマットで指示されたファイルとして(1度だけ)読み込みます。

• readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo) - 上の2つから内部的に呼ばれるメソッドです。指定されたfileInputFormatに基づいてpath内のファイルを読み込みます。指定されたwatchTypeに依存して、このソースは定期的に新しいデータのパス((FileProcessingMode.PROCESS_CONTINUOUSLY)を監視(各 interval ms) するか、現在パス内にあるデータを1度だけ処理して終了します(FileProcessingMode.PROCESS_ONCE)。pathFilterを使ってユーザは処理されるファイルを更に除外することができます。

  IMPLEMENTATION:

  背後では、Flinkはファイル読み込みプロセスを2つのサブタスク、すなわち directory monitoringと data readingに分割します。それぞれのサブタスクは個々のエンティティによって実装されます。監視は1つのnon-parallel(parallelism = 1)タスクによって実装されますが、読み込みは並行して実行する複数のタスクによって実行されます。後者の並行度はジョブの並行度に等しいです。1つの監視タスクの役割はディレクトリを走査(watchTypeに応じて定期的あるいは1度だけ)し、処理するファイルを見つけ、それらをsplitsし、それらの分割をダウンストリームのリーダーに割り当てます。リーダーは実際のデータを読み込むためのものです。各分割は1つのリーダーによって読み込まれますが、リーダーは1つずつ複数の分割を読むことができます。

  重要な注記:

  1. ファイルが修正された時に、もしwatchTypeがFileProcessingMode.PROCESS_CONTINUOUSLYに設定されると、その内容は完全に再処理されます。ファイルの最後に追加されるデータは再処理される全ての内容となるため、これは“確実に1回” セマンティクスを破壊します。

  2. もしwatchTypeがFileProcessingMode.PROCESS_ONCEに設定された場合、リーダーがファイルの内容の読み込みを完了するまで待たずに、ソースはパスを一回だけ走査し終了します。もちろん、リーダーは全てのファイルの内容が読み込まれるまで読み込みを続けます。ソースのクローズはその位置から後にチェックポイントが無いことに繋がります。ジョブは最後のチェックポイントから読み込みを再開するため、これはノードの障害の後での遅い回復に繋がるかもしれません。

ソケット ベース:

• socketTextStream - ソケットから読み込みます。要素はデリミタで区切ることができます。

コレクション ベース:

• fromCollection(Collection) - JavaのJava.util.Collectionからデータストリームを生成します。コレクション内の全ての要素は同じタイプでなければなりません。

• fromCollection(Iterator, Class) - イテレータからデータストリームを生成します。クラスはイテレータによって返される要素のデータタイプを指定します。

• fromElements(T ...) - オブジェクトの指定されたシーケンスからデータストリームを生成します。全てのオブジェクトは同じタイプでなければなりません。

• fromParallelCollection(SplittableIterator, Class) - イテレータから並列してデータストリームを生成します。クラスはイテレータによって返される要素のデータ型を指定します。

• generateSequence(from, to) - 並列して渡された間隔内の数値のシーケンスを生成します。

独自:

• addSource - 新しいソース関数をアタッチします。例えば、Aapache Kafkaから読み込むために、addSource(new FlinkKafkaConsumer<>(...))を使えます。詳細はconnectorsを参照してください。

ソースはプログラムが入力を取り込むところです。StreamExecutionEnvironment.addSource(sourceFunction)を使ってソースをプログラムに付け加えることができます。Flinkは多数の事前実装されたソース関数が同梱されますが、非並列ソースについてはSourceFunctionを実装することで独自のソースを常に書くことができ、並列ソースについてはParallelSourceFunctionインタフェースを実装するかRichParallelSourceFunctionを拡張することで独自のソースを常に書くことができます。

StreamExecutionEnvironmentからアクセス可能な幾つかの事前定義されたストリームソースがあります:

ファイルベース:

• readTextFile(path) - テキストファイル、つまりTextInputFormatの仕様を考慮するファイルを行ごとに読み込み、それらを文字列として返します。

• readFile(fileInputFormat, path) - 指定されたファイル入力フォーマットで指示されたファイルとして(1度だけ)読み込みます。

• readFile(fileInputFormat, path, watchType, interval, pathFilter) - これは上の2つから内部的に呼ばれるメソッドです。指定されたfileInputFormatに基づいてpath内のファイルを読み込みます。指定されたwatchTypeに依存して、このソースは定期的に新しいデータのパス((FileProcessingMode.PROCESS_CONTINUOUSLY)を監視(各 interval ms) するか、現在パス内にあるデータを1度だけ処理して終了します(FileProcessingMode.PROCESS_ONCE)。pathFilterを使ってユーザは処理されるファイルを更に除外することができます。

  IMPLEMENTATION:

  背後では、Flinkはファイル読み込みプロセスを2つのサブタスク、すなわち directory monitoringと data readingに分割します。それぞれのサブタスクは個々のエンティティによって実装されます。監視は1つのnon-parallel(parallelism = 1)タスクによって実装されますが、読み込みは並行して実行する複数のタスクによって実行されます。後者の並行度はジョブの並行度に等しいです。1つの監視タスクの役割はディレクトリを走査(watchTypeに応じて定期的あるいは1度だけ)し、処理するファイルを見つけ、それらをsplitsし、それらの分割をダウンストリームのリーダーに割り当てます。リーダーは実際のデータを読み込むためのものです。各分割は1つのリーダーによって読み込まれますが、リーダーは1つずつ複数の分割を読むことができます。

  重要な注記:

  1. ファイルが修正された時に、もしwatchTypeがFileProcessingMode.PROCESS_CONTINUOUSLYに設定されると、その内容は完全に再処理されます。ファイルの最後に追加されるデータは再処理される全ての内容となるため、これは“確実に1回” セマンティクスを破壊します。

  2. もしwatchTypeがFileProcessingMode.PROCESS_ONCEに設定された場合、リーダーがファイルの内容の読み込みを完了するまで待たずに、ソースはパスを一回だけ走査し終了します。もちろん、リーダーは全てのファイルの内容が読み込まれるまで読み込みを続けます。ソースのクローズはその位置から後にチェックポイントが無いことに繋がります。ジョブは最後のチェックポイントから読み込みを再開するため、これはノードの障害の後での遅い回復に繋がるかもしれません。

ソケット ベース:

• socketTextStream - ソケットから読み込みます。要素はデリミタで区切ることができます。

コレクション ベース:

• fromCollection(Seq) - JavaのJava.util.Collectionからデータストリームを生成します。コレクション内の全ての要素は同じタイプでなければなりません。

• fromCollection(Iterator) - イテレータからデータストリームを生成します。クラスはイテレータによって返される要素のデータタイプを指定します。

• fromElements(elements: _*) - 指定されたオブジェクトの系列からデータストリームを生成します。全てのオブジェクトは同じタイプでなければなりません。

• fromParallelCollection(SplittableIterator) - イテレータから並行してデータストリームを生成します。クラスはイテレータによって返される要素のデータ型を指定します。

• generateSequence(from, to) - 並列して渡された間隔内の数値のシーケンスを生成します。

独自:

• addSource - 新しいソース関数をアタッチします。例えば、Aapache Kafkaから読み込むために、addSource(new FlinkKafkaConsumer<>(...))を使えます。詳細はconnectorsを参照してください。

データシンクはDataStreamsを消費しそれらをファイル、ソケット、外部システムあるいは出力に転送します。 Flinkはデータストリーム上の操作の背後にカプセル化されている様々な組み込みの出力フォーマットが付属しています。

• writeAsText() / TextOutputFormat - 行ごとの要素を文字列として書き込みます。文字列はそれぞれの要素の*toString()*メソッドを呼ぶことで取得されます。

• writeAsCsv(...) / CsvOutputFormat - タプルをカンマ区切りの値のファイルとして書き込みます。行とフィールドのデリミタは設定可能です。各フィールドの値はオブジェクトの*toString()*メソッドから取得されます。

• print() / printToErr() - 各要素のtoString()値を標準出力/標準エラーストリームに出力します。オプションで、出力の先頭に付加されるプリフィックス(msg)を指定できます。これはprintの様々な呼び出しを区別するのに役立ちます。並列度が1より大きい場合、出力を生成したタスクの識別子が出力の前に追加されます。

• writeUsingOutputFormat() / FileOutputFormat - 独自のファイル出力のメソッドと基本クラス。独自のオブジェクトからバイトへの変換をサポートします。

• writeToSocket - SerializationSchemaに従ってソケットに要素を書き込みます

• addSink - 独自のシンク関数を呼び出します。Flinkにはシンク関数として実装される(Apache Kafkaのような)他のシステムへのコネクタがバンドルされています。

データシンクはDataStreamsを消費しそれらをファイル、ソケット、外部システムあるいは出力に転送します。 Flinkはデータストリーム上の操作の背後にカプセル化されている様々な組み込みの出力フォーマットが付属しています。

• writeAsText() / TextOutputFormat - 行ごとの要素を文字列として書き込みます。文字列はそれぞれの要素の*toString()*メソッドを呼ぶことで取得されます。

• writeAsCsv(...) / CsvOutputFormat - タプルをカンマ区切りの値のファイルとして書き込みます。行とフィールドのデリミタは設定可能です。各フィールドの値はオブジェクトの*toString()*メソッドから取得されます。

• print() / printToErr() - 各要素のtoString()値を標準出力/標準エラーストリームに出力します。オプションで、出力の先頭に付加されるプリフィックス(msg)を指定できます。これはprintの様々な呼び出しを区別するのに役立ちます。並列度が1より大きい場合、出力を生成したタスクの識別子が出力の前に追加されます。

• writeUsingOutputFormat() / FileOutputFormat - 独自のファイル出力のメソッドと基本クラス。独自のオブジェクトからバイトへの変換をサポートします。

• writeToSocket - SerializationSchemaに従ってソケットに要素を書き込みます

• addSink - 独自のシンク関数を呼び出します。Flinkにはシンク関数として実装される(Apache Kafkaのような)他のシステムへのコネクタがバンドルされています。