foreach_sink

@dp.foreach_sink() デコレータは ForEach シンクを定義します。これは、カスタム Python ロジックを使用してストリームを一度に 1 行ずつ処理します。データを書き込むには、追加フローまたは更新フローでシンクを target として参照します。概念的なガイダンス、考慮事項、および例については、パイプラインで各行を書き込むために ForEach シンクを使用するを参照してください。代わりに各マイクロバッチを DataFrame として処理するには、foreach_batch_sink を参照してください。

構文

関数からForEachシンクを定義するには、関数を@dp.foreach_sink()で装飾します。

Python

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引数にオブジェクトを渡します。

Python

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)）。

パラメーター

パラメーター	説明
行	必須。ストリームのレコードを1つ含む Row。関数、またはライターの process メソッドは、呼び出しごとに1行を受け取ります。
name	オプション。パイプライン内のシンクを識別する一意の名前。defaultは関数名です。ライターオブジェクトの場合、defaultは小文字のクラス名です。
関数	オプション。パイプラインが各行に対して呼び出す関数。デコレーターとしてではなく、直接 dp.foreach_sink() を呼び出す場合は、func を使用します。func と writer の両方を渡すことはできません。
ライター	オプションです。process(row)メソッド、およびオプションでopen(partition_id, epoch_id)とclose(error)メソッドを実装するオブジェクト。キーワード引数として渡します。funcとwriterの両方を渡すことはできません。

パラメーター	説明
行	必須。ストリームのレコードを1つ含む Row。関数、またはライターの process メソッドは、呼び出しごとに1行を受け取ります。
name	オプション。パイプライン内のシンクを識別する一意の名前。defaultは関数名です。ライターオブジェクトの場合、defaultは小文字のクラス名です。
関数	オプション。パイプラインが各行に対して呼び出す関数。デコレーターとしてではなく、直接 dp.foreach_sink() を呼び出す場合は、func を使用します。func と writer の両方を渡すことはできません。
ライター	オプションです。process(row)メソッド、およびオプションでopen(partition_id, epoch_id)とclose(error)メソッドを実装するオブジェクト。キーワード引数として渡します。funcとwriterの両方を渡すことはできません。

ライターメソッド

ライターオブジェクトは、Apache Spark Structured Streaming foreach ライターと同じメソッドを使用します。各マイクロバッチの各パーティションについて、パイプラインは open を呼び出し、次に open が True を返す場合は各行に対して process を呼び出し、最後に close を呼び出します。Structured Streaming でカスタム出力を書き込む方法に関する情報については、Use foreachBatch to write to arbitrary data sinks を参照してください。

手法	説明
open(partition_id, epoch_id)	オプション。パイプラインは、マイクロバッチのパーティションの行を処理する前に、このメソッドを呼び出します。これを使用して、接続を開くか、トランザクションを起動します。パーティションの行を処理するには True を返し、それらをスキップするには False を返します。ライターが open を定義していない場合、パイプラインは各パーティションの行を処理します。partition_id と epoch_id は、同じシンクに書き込むフローで、かつ完全更新の後で繰り返すことができます。それらを使用して行を重複排除する前に、失敗と重複行の処理を参照してください。
process(row)	必須。パイプラインはこのメソッドを行ごとに呼び出します。
close(error)	オプション。パイプラインは、パーティションの行を処理した後にこのメソッドを呼び出します。error は行の処理中に発生した例外であり、処理が成功した場合は None です。接続を閉じるために使用します。open がトランザクションを起動した場合は、error が None のときのみ commit し、それ以外の場合はロールバックします。

手法	説明
open(partition_id, epoch_id)	オプション。パイプラインは、マイクロバッチのパーティションの行を処理する前に、このメソッドを呼び出します。これを使用して、接続を開くか、トランザクションを起動します。パーティションの行を処理するには True を返し、それらをスキップするには False を返します。ライターが open を定義していない場合、パイプラインは各パーティションの行を処理します。partition_id と epoch_id は、同じシンクに書き込むフローで、かつ完全更新の後で繰り返すことができます。それらを使用して行を重複排除する前に、失敗と重複行の処理を参照してください。
process(row)	必須。パイプラインはこのメソッドを行ごとに呼び出します。
close(error)	オプション。パイプラインは、パーティションの行を処理した後にこのメソッドを呼び出します。error は行の処理中に発生した例外であり、処理が成功した場合は None です。接続を閉じるために使用します。open がトランザクションを起動した場合は、error が None のときのみ commit し、それ以外の場合はロールバックします。