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

ソースはプログラムが入力を取り込むところです。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を使ってユーザは処理されるファイルを更に除外することができます。

  実装:

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

  重要な注記:

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

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

ソケット ベース:

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

コレクション ベース:

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

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

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

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

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

独自:

• addSource - 新しいソース関数を付け加えます。例えば、Apache Kafkaから読み込むために、addSource(new FlinkKafkaConsumer08<>(...))を使うことができます。詳細はコネクタ を見てください。

ソースはプログラムが入力を取り込むところです。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を使ってユーザは処理されるファイルを更に除外することができます。

  実装:

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

  重要な注記:

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

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

ソケット ベース:

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

コレクション ベース:

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

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

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

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

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

独自:

• addSource - 新しいソース関数を付け加えます。例えば、Apache Kafkaから読み込むために、addSource(new FlinkKafkaConsumer08<>(...))を使うことができます。詳細はコネクタ を見てください。

データシンクは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のような)他のシステムへのコネクタと一緒にまとめられています。

繰り返しのストリーミングプログラムはステップ関数を実装し、それをIterativeStreamに埋め込みます。DataStream プログラムは終了しないかもしれないため、繰り返しの最大数はありません。その代わり、split 変換あるいは filter を使って、ストリームのどの部分が繰り返しにフィードバックされるか、そしてどの部分がダウンストリームに転送されるかを指定する必要があります。ここで、フィルターを使った例を示します。まず、IterativeStreamを定義します。

IterativeStream<Integer> iteration = input.iterate();

次に、変換(ここでは単純なmap 変換です)の連続を使ってループ内で実行されるロジックを指定します

DataStream<Integer> iterationBody = iteration.map(/* this is executed many times */);

繰り返しを閉じ、繰り返しの最後を定義するために、IterativeStreamのcloseWith(feedbackStream)メソッドを呼びます。closeWith 関数に指定されたDataStreamは繰り返しの頭にフィードバックされるでしょう。一般的なパターンは、フィードバックされるストリームの一部と前方に広められるストリームの一部を分割するために、フィルターを使います。これらのフィルターは、例えば要素がフィードバックされずにダウンストリームに広まることができる “termination” ロジックを定義することができます。

iteration.closeWith(iterationBody.filter(/* one part of the stream */));
DataStream<Integer> output = iterationBody.filter(/* some other part of the stream */);

デフォルトでは、フィードバックストリームのパーティション化はイテレーションの頭の入力と同じように自動的に設定されるでしょう。これを上書きするために、ユーザはcloseWith内で任意のbooleanフラグを設定することができます。

例えば、これは整数の連続からそれらが0になるまで1を継続的に引くプログラムです:

DataStream<Long> someIntegers = env.generateSequence(0, 1000);

IterativeStream<Long> iteration = someIntegers.iterate();

DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
  @Override
  public Long map(Long value) throws Exception {
    return value - 1 ;
  }
});

DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
  @Override
  public boolean filter(Long value) throws Exception {
    return (value > 0);
  }
});

iteration.closeWith(stillGreaterThanZero);

DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
  @Override
  public boolean filter(Long value) throws Exception {
    return (value <= 0);
  }
});

繰り返しのストリーミングプログラムはステップ関数を実装し、それをIterativeStreamに埋め込みます。DataStream プログラムは終了しないかもしれないため、繰り返しの最大数はありません。その代わり、split 変換あるいは filter を使って、ストリームのどの部分が繰り返しにフィードバックされるか、そしてどの部分がダウンストリームに転送されるかを指定する必要があります。ここで、ボディ(繰り返される計算の一部分)が単純なマップ変換の繰り返しの例を示します。フィードバックされる要素はフィルタを使ってダウンロードストリームに転送される要素と識別されます。

val iteratedStream = someDataStream.iterate(
  iteration => {
    val iterationBody = iteration.map(/* this is executed many times */)
    (tail.filter(/* one part of the stream */), tail.filter(/* some other part of the stream */))
})

デフォルトでは、フィードバックストリームのパーティション化はイテレーションの頭の入力と同じように自動的に設定されるでしょう。これを上書きするために、ユーザはcloseWith内で任意のbooleanフラグを設定することができます。

例えば、これは整数の連続からそれらが0になるまで1を継続的に引くプログラムです:

val someIntegers: DataStream[Long] = env.generateSequence(0, 1000)

val iteratedStream = someIntegers.iterate(
  iteration => {
    val minusOne = iteration.map( v => v - 1)
    val stillGreaterThanZero = minusOne.filter (_ > 0)
    val lessThanZero = minusOne.filter(_ <= 0)
    (stillGreaterThanZero, lessThanZero)
  }
)