append_flow

@dp.append_flowデコレーターは、パイプラインテーブルの追加フローまたはバックフィルを作成します。この関数は、Apache SparkストリーミングDataFrameを返す必要があります。LakeFlowパイプラインフローを使用してデータを段階的にロードおよび処理するを参照してください。

追加フローは、ストリーミングテーブル、マネージドテーブル、またはシンクをターゲットにすることができます。

構文

Python
from pyspark import pipelines as dp

dp.create_streaming_table("<target-table-name>") # Required only if the target table doesn't exist.

@dp.append_flow(
  target = "<target-table-name>",
  name = "<flow-name>", # optional, defaults to function name
  once = False, # optional
  depends_on = "<flow-name>", # optional, Public Preview
  spark_conf = {"<key>" : "<value", "<key" : "<value>"}, # optional
  comment = "<comment>", # optional
  import_checkpoint = "<checkpoint-path>") # optional
def <function-name>():
  return (<streaming-query>) #

パラメーター

パラメーター	Type	説明
function	`function`	必須。ユーザー定義のクエリから Apache Sparkストリーミングデータフレームを返す関数。
`target`	`str`	必須。追加フローのターゲットとなるテーブルまたはシンクの名前。
`name`	`str`	フロー名。指定されていない場合は、デフォルトで関数名になります。
`once`	`bool`	必要に応じて、フローをバックフィルなどの 1 回限りのフローとして定義します。`once=True`を使用すると、フローは次の 2 つの方法で変化します。戻り値。`streaming-query` 。この場合、ストリーミング DataFrame ではなく、バッチ DataFrame である必要があります。デフォルトでは、フローは 1 回実行されます。パイプラインが完全リフレッシュで更新されると、 `ONCE`フローが再度実行され、データが再作成されます。
`depends_on`	`str` または `list`	パブリックプレビュー。このフローが起動される前に正常に完了する必要がある1つ以上のフロー名。単一のフロー名または名前のリストを受け入れます。連続パイプラインでは、各先行するフローは1回限りの（ `once` ）フローである必要があります。これはフローの実行順序のみを決定するものであり、フローの実行方法は変更しません。「depends_onを使用してパイプラインフローの実行順序を決定する」を参照してください。
`comment`	`str`	フローの説明。
`spark_conf`	`dict`	このクエリを実行するためのSpark構成のリスト
`import_checkpoint`	`str`	フローにインポートする既存の Structured Streaming チェックポイントへのパス。これにより、移行されたストリームはソースの再処理を行う代わりに、最後にコミットされたオフセットから再開されます。チェックポイントのインポートはベータ版です。Structured Streaming チェックポイントの移行を参照してください。

パラメーター	Type	説明
function	`function`	必須。ユーザー定義のクエリから Apache Sparkストリーミングデータフレームを返す関数。
`target`	`str`	必須。追加フローのターゲットとなるテーブルまたはシンクの名前。
`name`	`str`	フロー名。指定されていない場合は、デフォルトで関数名になります。
`once`	`bool`	必要に応じて、フローをバックフィルなどの 1 回限りのフローとして定義します。`once=True`を使用すると、フローは次の 2 つの方法で変化します。戻り値。`streaming-query` 。この場合、ストリーミング DataFrame ではなく、バッチ DataFrame である必要があります。デフォルトでは、フローは 1 回実行されます。パイプラインが完全リフレッシュで更新されると、 `ONCE`フローが再度実行され、データが再作成されます。
`depends_on`	`str` または `list`	パブリックプレビュー。このフローが起動される前に正常に完了する必要がある1つ以上のフロー名。単一のフロー名または名前のリストを受け入れます。連続パイプラインでは、各先行するフローは1回限りの（ `once` ）フローである必要があります。これはフローの実行順序のみを決定するものであり、フローの実行方法は変更しません。「depends_onを使用してパイプラインフローの実行順序を決定する」を参照してください。
`comment`	`str`	フローの説明。
`spark_conf`	`dict`	このクエリを実行するためのSpark構成のリスト
`import_checkpoint`	`str`	フローにインポートする既存の Structured Streaming チェックポイントへのパス。これにより、移行されたストリームはソースの再処理を行う代わりに、最後にコミットされたオフセットから再開されます。チェックポイントのインポートはベータ版です。Structured Streaming チェックポイントの移行を参照してください。

例

Python
from pyspark import pipelines as dp

# Create a sink for an external Delta table
dp.create_sink("my_sink", "delta", {"path": "/tmp/delta_sink"})

# Add an append flow to an external Delta table
@dp.append_flow(name = "flow", target = "my_sink")
def flowFunc():
  return <streaming-query>

# Add a backfill
@dp.append_flow(name = "backfill", target = "my_sink", once = True)
def backfillFlowFunc():
    return (
      spark.read
      .format("json")
      .load("/path/to/backfill/")
    )

# Create a Kafka sink
dp.create_sink(
  "my_kafka_sink",
  "kafka",
  {
    "kafka.bootstrap.servers": "host:port",
    "topic": "my_topic"
  }
)

# Add an append flow to a Kafka sink
@dp.append_flow(name = "flow", target = "my_kafka_sink")
def myFlow():
  return read_stream("xxx").select(F.to_json(F.struct("*")).alias("value"))

Structured Streaming チェックポイントの移行

備考

ベータ版

チェックポイントのインポートはベータ版です。

ソースを再処理せずに、既存のStructured Streamingワークロードをパイプラインに移行するには、import_checkpointを使用します。Structured Streamingクエリーで使用したcheckpointLocationに設定します。には、クラウドストレージ、Unity Catalogボリューム、またはDBFSパスを指定できます。パイプラインの初回更新時に、フローはそのチェックポイントをパイプラインの管理対象ストレージに複製します。その後、フローは集計、重複排除キー、ウォーターマークなどの状態を保持したまま、最後にコミットされたオフセットから再開します。後続のパイプライン更新では、フローは複製されたチェックポイントを使用します。元のチェックポイントは変更されません。

フローのターゲットは、create_table またはシンクで作成されたマネージドテーブルである必要があります。

警告

パイプラインを実行する前に、元の Structured Streaming クエリーを停止します。元の Structured Streaming クエリーはインポート後に再利用できますが、そのチェックポイント状態を管理し、パイプラインとクエリーが同時に同じテーブルに書き込まないようにする必要があります（書き込むとデータが重複する可能性があります）。

Structured Streaming クエリーを、クエリーが書き込んだテーブルへの書き込みを継続するパイプラインフローとして再作成し、そのチェックポイントをインポートします。

Python
from pyspark import pipelines as dp

# Use the existing table your Structured Streaming query writes to; create_table adopts it.
dp.create_table("streaming_table")

# Continue from the imported checkpoint instead of reprocessing the source.
@dp.append_flow(
  target = "streaming_table",
  import_checkpoint = "/Volumes/my_catalog/my_schema/checkpoints/my_stream",
)
def migrate():
  # The same source your original query read from.
  return spark.readStream.table("source_table")

チェックポイントは、最初のパイプライン更新時にのみインポートされます。以降の更新ではimport_checkpointは無視されます。完全更新では、チェックポイントは再インポートされません。新しい空のチェックポイントから起動され、ソースが再処理されます。別のチェックポイントをインポートするには、ターゲットテーブルでこれまで使用されていないフロー名を使用してください。既存のフロー名を再利用すると、インポートはスキップされます。

制限事項

import_checkpoint は append_flow でのみサポートされています。

構文​

パラメーター​

例​

Structured Streaming チェックポイントの移行​

制限事項​

構文

パラメーター

例

Structured Streaming チェックポイントの移行

制限事項