ForEach シンクを使用してパイプラインの各行を書き込みます

ForEach シンクを使用して、外部 REST API やキー値ストアなど、組み込みシンクがない宛先にストリーミングデータを書き込みます。

ForEach シンクは、一度に 1 行ずつストリームを処理します。各行に対して、パイプラインは、Apache Spark Structured Streaming の foreach シンクと同様に、カスタム Python 関数またはライターオブジェクトを呼び出します。

注記

このページで説明されている ForEach シンクは、パイプライン用です。Apache Spark Structured Streamingはforeachもサポートしています。Structured Streamingでカスタム出力を書き込む方法に関する情報については、foreachBatchを使用して任意のデータシンクに書き込むを参照してください。

ForEachシンクを使用するタイミング

外部 API の呼び出しや各レコードのキーバリューストアの更新など、組み込みのシンク形式がない宛先に各レコードを個別に書き込む必要がある場合は、ForEach シンクを使用します。ロジックが Delta Lake テーブルへのマージや DataFrame 操作による複数の宛先への書き込みなど、マイクロバッチ全体で機能する場合は、代わりに ForEachBatch シンクを使用してください。

この表は、2つのシンクを比較しています：

characteristic	ForEach シンク	ForEachBatch シンク
処理の単位	一度に1行	一度に1つのマイクロバッチDataFrame
コードの実行場所	エグゼキューターで	ドライバー上
Python API	@dp.foreach_sink()	@dp.foreach_batch_sink()

characteristic	ForEach シンク	ForEachBatch シンク
処理の単位	一度に1行	一度に1つのマイクロバッチDataFrame
コードの実行場所	エグゼキューターで	ドライバー上
Python API	@dp.foreach_sink()	@dp.foreach_batch_sink()

組み込みシンクの情報については、LakeFlow Pipelinesのシンクを参照してください。@dp.foreach_sink() Python API リファレンスについては、foreach_sink を参照してください。

ForEach シンクの定義

関数またはライターオブジェクトからForEachシンクを定義できます。

関数からシンクを定義するには、関数に@dp.foreach_sink()を適用します。パイプラインは、各行に対して関数を呼び出します。name を省略した場合、シンクは関数名を使用します：

Python

from pyspark import pipelines as dp

@dp.foreach_sink()
def my_row_sink(row):
    # Custom per-row logic here.
    ...

ライター オブジェクトからシンクを定義するには、writer キーワード引数を使用してオブジェクトを渡します。ライターは process(row) を実装する必要があり、open(partition_id, epoch_id) と close(error) を実装できます。各パーティションに対して 1 つの接続を開き、行ごとに再利用する場合にライターを使用します。name を省略した場合、シンクはライターのクラス名を小文字で使用します。

Python

from pyspark import pipelines as dp

class MyRowWriter:
    def open(self, partition_id, epoch_id):
        # Open a connection for this partition.
        return True

    def process(self, row):
        # Write the row by using the connection.
        ...

    def close(self, error):
        # Close the connection.
        ...

dp.foreach_sink(name="my_row_sink", writer=MyRowWriter())

各メソッドの詳細については、Writerメソッドを参照してください。

イベントログエントリ

ForEach シンクを定義すると、パイプラインイベントログは、そのシンクに対して "format": "foreach" を持つ SinkDefinition イベントを記録します。

ForEach シンクへの書き込み

追加フローまたは更新フローから ForEach シンクに書き込むことができます。

追加フローで書き込み

追加フローから ForEach シンクに書き込むには、シンク名を target として参照します。

Python

@dp.append_flow(target="my_row_sink")
def my_flow():
    return spark.readStream.table("my_catalog.my_schema.source_table")

同じForEachシンクを対象とする複数の追加フローを定義できます。各フローは独自のチェックポイントを保持します。

更新フローで書き込む

更新フローからForEachシンクに書き込むには、@dp.update_flowデコレータを使用します。

Python

@dp.update_flow(target="my_row_sink")
def my_stateful_flow():
    return spark.readStream.table("my_catalog.my_schema.events").groupBy("user_id").count()

更新フローでは、後続のマイクロバッチで新しい値とともに同じキーをシンクに再度送信できます。古い値を置き換えるには、挿入する代わりにキーで各行を宛先にアップサートします。前の例では、キーとして user_id を使用して、ユーザーの新しいカウントが前のカウントを置き換えるようにします。

低レイテンシー処理の場合、ForEach シンクに書き込む更新フローをリアルタイムモードで実行できます。LakeFlow Pipelinesでリアルタイムモードを使用するを参照してください。

エグゼキューターでライブラリと依存関係をインポートする

関数またはライターはドライバー上ではなく、エグゼキューター上で実行されます。次のプラクティスに従ってください。

• コードに必要なパッケージをパイプライン環境に追加します。パイプライン環境のパッケージは、エグゼキューター上のコードから利用できます。パイプラインの Python 依存関係の管理を参照してください。
• コードが関数またはライターメソッド内で使用するライブラリは、パイプラインソースファイルの先頭だけでなく、インポートしてください。
• ファイルの先頭で開くクライアントや接続など、ドライバー上で作成するオブジェクトに依存しないでください。関数内、またはライターのopenメソッド内で作成してください。

障害と重複行の処理

コードで例外が発生すると、フローは失敗し、パイプラインのイベントログに例外メッセージを含むエラーイベントが記録されます。

マイクロバッチが失敗した後に再試行されると、シンクが再度いくつかの行を受け取ることがあります。例えば、行のユニークキーを使用して宛先にアップサートするなどして、行ごとの書き込みを冪等にするようにします。LakeFlow Pipelinesでの処理保証を参照してください。

ライター オブジェクトを使用する場合、open は partition_id と epoch_id を受け取ります。これらを合わせると、1つのマイクロバッチ内の1つのパーティションの行を識別できるため、それを使用して行の重複を排除できます。関数はこれらのIDを受け取りません。

IDは、1つのフローのチェックポイント内でのみ一意です。同じペアは、以下の場合に繰り返すことができます。

• 複数のフローが同じシンクに書き込みます。
• 完全更新はチェックポイントをResetします。

これらのIDで重複を排除するには、単一のフローからシンクに書き込み、完全更新を実行するときに重複排除の状態をクリアしてください。

フルリフレッシュ

完全更新時に、パイプラインは ForEach シンクに書き込む各フローのチェックポイントをResetします。パイプラインはシンクが書き込むデータを追跡またはクリーンアップしないため、宛先の保持とクリーンアップはユーザーが管理します。クリーンな状態からやり直す必要がある場合は、そのデータを自分で削除してください。

制限

• 行メトリクス ：パイプラインは、ForEach シンクが書き込む行数をレポートしません。
• 行ごとの呼び出し : コードが行ごとに外部システムを呼び出すと、外部システムの throughput が制限される可能性があります。オーバーヘッドを削減するには、ライターの open メソッドで接続を開いて各行で再利用するか、ForEachBatch シンクを使用して一括で書き込みます。

例

これらの例は、ForEach シンクを使用する一般的な方法を示しています。

基本構文の例

この例では、関数から ForEach シンクを定義し、追加フローからそこに書き込みます。

Python

from pyspark import pipelines as dp

# Create a ForEach sink
@dp.foreach_sink(name="my_foreach_sink")
def handle_row(row):
    # Custom per-row logic here.
    ...

# Create source data for the example
@dp.table()
def example_source_data():
    return spark.range(5)

# Write to the sink from an append flow
@dp.append_flow(target="my_foreach_sink")
def my_flow():
    return spark.readStream.table("example_source_data")

各行を外部APIに書き込む

この例では、ライターオブジェクトを使用して各オーダーを外部 REST API に送信します。ライターはパーティションごとに 1 つの HTTP セッションを開きます。URL 内の注文 ID を使用するため、再試行された行は重複を作成するのではなく、同じレコードを更新します。

Python

from pyspark import pipelines as dp

class OrdersApiWriter:
    def open(self, partition_id, epoch_id):
        import requests  # Import on the executor

        self.session = requests.Session()
        return True

    def process(self, row):
        response = self.session.put(
            f"https://api.example.com/orders/{row.order_id}",
            json={"order_id": row.order_id, "status": row.status},
            timeout=10,
        )
        # Raise on HTTP errors so that the flow fails instead of dropping the row.
        response.raise_for_status()

    def close(self, error):
        self.session.close()

dp.foreach_sink(name="orders_api_sink", writer=OrdersApiWriter())

@dp.append_flow(target="orders_api_sink")
def orders_to_api():
    return spark.readStream.table("my_catalog.my_schema.orders")

複数のフローから単一のシンクに書き込む

この例では、2つのソースを同じForEachシンクに書き込みます。

Python

from pyspark import pipelines as dp

@dp.foreach_sink(name="events_sink")
def handle_event(row):
    # Custom per-row logic here.
    ...

@dp.append_flow(target="events_sink")
def web_events():
    return spark.readStream.table("my_catalog.my_schema.web_events")

@dp.append_flow(target="events_sink")
def mobile_events():
    return spark.readStream.table("my_catalog.my_schema.mobile_events")

その他のリソース

• foreach_sink
• パイプラインの任意のデータシンクに書き込むために ForEachBatch を使用します
• foreachBatch を使用して任意のデータ シンクに書き込む
• LakeFlow Pipelines の処理保証
• LakeFlow Pipelinesでリアルタイムモードを使用する