append_flow
@dlt.append_flow
デコレーターは、DLT テーブルの追加フローまたはバックフィルを作成します。この関数は、 Apache Sparkストリーミング DataFrameを返す必要があります。 DLT フローを使用したデータの増分読み込みと処理を参照してください。
追加フローは、ストリーミングテーブルまたはシンクをターゲットにできます。
構文
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
spark_conf = {"<key>" : "<value", "<key" : "<value>"}, # optional
comment = "<comment>") # optional
def <function-name>():
return (<streaming query>)
パラメーター
パラメーター | タイプ | 説明 |
---|---|---|
関数 |
| 必須。ユーザー定義のクエリから Apache Sparkストリーミング DataFrame を返す関数。 |
|
| 必須。追加フローのターゲットであるテーブルまたはシンクの名前。 |
|
| フロー名。指定しない場合、デフォルトは関数名になります。 |
|
| フローの説明。 |
|
| このクエリを実行するための 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>
# 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"))