replace_flow

Important

這項功能位於 測試版 (Beta) 中。

裝飾器會 @dp.replace_flow 為你的管線中的串流表建立一個替換使用流程。 每次更新時，流程會替換目標資料表中所有與關鍵欄位相符 replace_using 的列，其他列則保持不變。 函式必須傳回 Apache Spark 串流 DataFrame。 請參見部分快照替換，並用替換使用流程。

當你的來源是一系列以欄位鍵入的部分快照時，請使用 @dp.replace_flow 。 若要在單一語句中定義目標資料表和流程，請將 傳遞 replace_using 至 sequence_by@dp.table。

Syntax

from pyspark import pipelines as dp

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

@dp.replace_flow(
  target = "<target-table-name>",
  replace_using = ["<key-column>", "<key-column>"],
  sequence_by = "<sequence-column>",
  name = "<flow-name>", # optional, defaults to function name
  comment = "<comment>", # optional
  spark_conf = {"<key>" : "<value>", "<key>" : "<value>"}) # optional
def <function-name>():
  return (<streaming-query>)

參數

參數 類型 Description
函式 function Required. 用來從使用者定義的查詢中返回 Apache Spark 串流 DataFrame 的函數。
target str Required. 流量目標的串流表名稱。
replace_using list Required. 關鍵欄位用來標示要替換哪些目標列。 至少指定一欄。 鍵欄位不能重複，且每個鍵欄位的類型必須是可排序的。
sequence_by str 或 Column Required. 就是那個下單更新的欄位。 對於每個鍵，最高序列勝出，且較低序列的列永遠不會覆蓋目標中已存在的較高序列。
name str 流程名稱。 如果未提供，則預設為函式名稱。
comment str 對流程的描述。
spark_conf dict 這是執行此查詢時的 Spark 配置清單。

Examples

from pyspark import pipelines as dp

# Keep the latest row for each order from a stream of partial snapshots
dp.create_streaming_table("orders_current")

@dp.replace_flow(
  target = "orders_current",
  replace_using = ["order_id"],
  sequence_by = "updated_at"
)
def orders_flow():
  return spark.readStream.table("order_updates")

當紀錄由多個欄位組合識別時，請使用多於一個金鑰欄位：

from pyspark import pipelines as dp

dp.create_streaming_table("accounts_current")

@dp.replace_flow(
  target = "accounts_current",
  replace_using = ["region", "account_id"],
  sequence_by = "updated_at"
)
def accounts_flow():
  return spark.readStream.table("account_updates")

Limitations

  • 串流表支援單一REPLACE USING流程，無法與其他流程類型結合REPLACE USING，例如附加流程、自動 CDC 流程或流程。REPLACE WHERE
  • 查詢必須是串流查詢。 @dp.replace_flow 拒絕非串流來源。
  • REPLACE USING 流程需要 Databricks Runtime 18.2 及以上版本。
  • Last updated on