foreach_sink

@dp.foreach_sink()デコレータはForEachシンクを定義し、カスタムPythonロジックで1行ずつストリームを処理する。 データを書き込むために、アペンドフローや更新フローの中のtargetとしてシンクを参照します。 概念的な指針、考慮事項、例については「 パイプラインの各行を書くために ForEach シンクを使う」を参照してください。 各マイクロバッチをデータフレームとして処理する場合は、 foreach_batch_sinkを参照してください。

Syntax

関数から ForEach シンクを定義するには、関数を次のように装飾します @dp.foreach_sink():

from pyspark import pipelines as dp

@dp.foreach_sink(name="<name>")
def row_handler(row):
    """
    Required:
      - `row`: a Row that contains one record of the stream.
    """
    # Your custom write logic here.
    # This function runs on the executors, so import any libraries it uses inside the function.

ライターオブジェクトから ForEach シンクを定義するには、 dp.foreach_sink() を呼び出して writer 引数でオブジェクトを渡します:

from pyspark import pipelines as dp

class RowWriter:
    def open(self, partition_id, epoch_id):
        # Optional. Return False to skip the rows of this partition.
        return True

    def process(self, row):
        # Required. Called for each row.
        ...

    def close(self, error):
        # Optional. `error` is None if processing succeeded.
        ...

dp.foreach_sink(name="<name>", writer=RowWriter())

デコレーターを使う代わりにfunc引数でdp.foreach_sink()に関数を渡すこともできます。例えばdp.foreach_sink(name="<name>", func=row_handler)。

Parameters

Parameter Description
行 Required. ストリームの記録が1つだけ入った Row 。 あなたの関数、つまりライターの process メソッドは、1コールごとに1行を受け取ります。
name Optional. パイプライン内のシンクを識別する一意の名前。 関数名をデフォルトで使います。 ライターオブジェクトの場合、デフォルトで小文字のクラス名が使われます。
func Optional. パイプラインが各行ごとに呼び出す関数です。 デコレーターとしてではなく、直接dp.foreach_sink()に電話するときは、そのfuncを使いましょう。 funcとwriterの両方に合格することはできません。
writer Optional. process(row)メソッドを実装し、オプションでopen(partition_id, epoch_id)メソッドとclose(error)メソッドを実装するオブジェクトです。 キーワードの引数として渡してください。 funcとwriterの両方に合格することはできません。

ライターメソッド

ライターオブジェクトは、Apache Sparkの構造化ストリーミング foreach ライターと同じメソッドを使用します。 各マイクロバッチの各パーティションについて、パイプラインはopenを呼び出し、openがTrue返した場合は各行ごとにprocessを呼び出し、最後に closeを呼び出します。 構造化ストリーミングでカスタム出力を書く方法については、「 foreachBatchを使って任意のデータシンクに書き込み」をご覧ください。

Method Description
open(partition_id, epoch_id) Optional. パイプラインはこのメソッドを呼び出し、マイクロバッチのパーティションの行を処理する前に行います。 接続を開いたり取引を始めたりするために使うのが良いでしょう。 パーティションの行を処理するには True を戻すか、スキップするには False 。 もし書き手が openを定義していない場合、パイプラインは各パーティションの行を処理します。 partition_id また、同じシンクに書き込むフロー間で epoch_id が完全にリフレッシュされた後に繰り返し書き込むことができます。 重複を減らす前に、「 失敗処理」と「重複行の処理」を参照してください。
process(row) Required. パイプラインは各行ごとにこのメソッドを呼び出します。
close(error) Optional. パイプラインはパーティションの行を処理する後にこのメソッドを呼び出します。 error は行処理中に発生した例外であり、処理が成功した場合は None です。 それを使って接続を閉じましょう。 トランザクションopen開始した場合は、トランザクションがNoneerror時にのみコミットし、そうでなければロールバックします。
  • 最終更新: