メインコンテンツまでスキップ

append_flow

@dlt.append_flow デコレーターは、Lakeflow 宣言型パイプライン テーブルの追加フローまたはバックフィルを作成します。この関数は、 Apache Sparkストリーミング DataFrameを返す必要があります。 「宣言型パイプライン フローを使用したデータの増分読み込みと処理 」を参照してくださいLakeflow

追加フローは、ストリーミングテーブルまたはシンクをターゲットにできます。

構文

Python
import dlt

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

@dlt.append_flow(
target = "<target-table-name>",
name = "<flow-name>", # optional, defaults to function name
once = <boolean>, # optional, defaults to false
spark_conf = {"<key>" : "<value", "<key" : "<value>"}, # optional
comment = "<comment>") # optional
def <function-name>():
return (<streaming-query>) #

パラメーター

パラメーター

タイプ

説明

function

function

必須。ユーザー定義のクエリから Apache Sparkストリーミング データフレーム を返す関数。

target

str

必須。追加フローのターゲットであるテーブルまたはシンクの名前。

name

str

フロー名。指定しない場合、デフォルトは関数名になります。

once

bool

必要に応じて、フローをバックフィルなどの 1 回限りのフローとして定義します。once=Trueを使用すると、フローが次の 2 つの方法で変更されます。

  • 戻り値。streaming-query。この場合は、ストリーミング DataFrame ではなく、バッチ DataFrame である必要があります。
  • フローはデフォルトで 1 回実行されます。パイプラインが完全に更新されて更新されると、 ONCE フローが再度実行されてデータが再作成されます。

comment

str

フローの説明。

spark_conf

dict

このクエリを実行するための Spark 構成の一覧

Python
import dlt

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

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

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

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

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