Delta Live Tables Python 語言參考

本文提供差異即時資料表 Python 程式設計介面的詳細資料。

如需 SQL API 的相關資訊，請參閱差異即時資料表 SQL 語言參考。

如需設定自動載入器的特定詳細資料，請參閱什麼是自動載入器？。

開始之前

當您使用差異即時資料表Python 介面實作管線時，以下是重要的考量：

• 由於 Python table() 和 view() 函式會在管線更新的規劃和執行期間多次叫用，因此請勿在其中一個可能有副作用的函式中包含程式碼 (例如，修改資料或傳送電子郵件的程式碼)。 為了避免非預期的行為，定義資料集的 Python 函式應該只包含定義資料表或檢視所需的程式碼。
• 若要執行像是傳送電子郵件或與外部監視服務整合等作業，特別是在定義資料集的函式中，請使用事件勾點。 在定義資料集的函式中實作這些作業會導致非預期的行為。
• Python table 和 view 函式必須傳回 DataFrame。 某些在 DataFrame 上運作的函式不會傳回 DataFrame，而且不應該使用。 這些作業包括 collect()、count()、toPandas()、save() 和 saveAsTable() 等函式。 由於 DataFrame 轉換會在解析完整資料流程圖形之後執行，因此使用這類作業可能會有非預期的副作用。

匯入 dlt Python 模組

差異即時資料表 Python 函式定義於 dlt 模組中。 使用 Python API 實作的管線必須匯入此課程模組：

import dlt

建立差異即時資料表具體化檢視或串流資料表

在 Python 中，差異即時資料表會根據定義查詢來決定將資料集更新為具體化檢視或串流資料表。 @table 裝飾項目可用來定義具體化檢視和串流資料表。

若要在 Python 中定義具體化檢視，請將 @table 套用至對資料來源執行靜態讀取的查詢。 若要定義串流資料表，請將 @table 套用至對資料來源執行串流讀取的查詢，或使用 create_streaming_table () 函式。 這兩種資料集類型都有相同的語法規格，如下所示：

import dlt

@dlt.table(
  name="<name>",
  comment="<comment>",
  spark_conf={"<key>" : "<value>", "<key>" : "<value>"},
  table_properties={"<key>" : "<value>", "<key>" : "<value>"},
  path="<storage-location-path>",
  partition_cols=["<partition-column>", "<partition-column>"],
  cluster_by = ["<clustering-column>", "<clustering-column>"],
  schema="schema-definition",
  row_filter = "row-filter-clause",
  temporary=False)
@dlt.expect
@dlt.expect_or_fail
@dlt.expect_or_drop
@dlt.expect_all
@dlt.expect_all_or_drop
@dlt.expect_all_or_fail
def <function-name>():
    return (<query>)

建立差異即時資料表檢視

若要在 Python 中定義檢視，請套用 @view 裝飾項目。 @table如同裝飾項目，您可以在差異即時資料表中針對靜態或串流資料集使用檢視。 以下是使用 Python 定義檢視的語法：

import dlt

@dlt.view(
  name="<name>",
  comment="<comment>")
@dlt.expect
@dlt.expect_or_fail
@dlt.expect_or_drop
@dlt.expect_all
@dlt.expect_all_or_drop
@dlt.expect_all_or_fail
def <function-name>():
    return (<query>)

定義資料表和檢視

若要在 Python 中定義資料表或檢視，請將 @dlt.view 或 @dlt.table 裝飾項目套用至函式。 您可以使用函式名稱或 name 參數來指派資料表或檢視名稱。 下列範例會定義兩個不同的資料集：一個稱為 taxi_raw 的檢視，其接受 JSON 檔案做為輸入來源，而名為 filtered_data 的資料表會接受 taxi_raw 檢視做為輸入：

import dlt

@dlt.view
def taxi_raw():
  return spark.read.format("json").load("/databricks-datasets/nyctaxi/sample/json/")

# Use the function name as the table name
@dlt.table
def filtered_data():
  return dlt.read("taxi_raw").where(...)

# Use the name parameter as the table name
@dlt.table(
  name="filtered_data")
def create_filtered_data():
  return dlt.read("taxi_raw").where(...)

範例：存取在相同管線中定義的資料集

除了從外部資料來源讀取之外，您還可以使用差異即時資料表 read() 函式存取相同管線中定義的資料集。 下列範例示範如何使用 read() 函式建立 customers_filtered 資料集：

@dlt.table
def customers_raw():
  return spark.read.format("csv").load("/data/customers.csv")

@dlt.table
def customers_filteredA():
  return dlt.read("customers_raw").where(...)

您也可以使用 spark.table() 函式來存取相同管線中定義的資料集。 使用 spark.table() 函式來存取管線中定義的資料集時，在函式引數前面加上 LIVE 關鍵字至資料集名稱：

@dlt.table
def customers_raw():
  return spark.read.format("csv").load("/data/customers.csv")

@dlt.table
def customers_filteredB():
  return spark.table("LIVE.customers_raw").where(...)

範例：從中繼存放區中註冊的資料表讀取

若要從 Hive 中繼存放區中註冊的資料表讀取資料，請在函式引數中省略 LIVE 關鍵字，並選擇性地使用資料庫名稱限定資料表名稱：

@dlt.table
def customers():
  return spark.table("sales.customers").where(...)

如需從 Unity 目錄資料表讀取的範例，請參閱將資料內嵌至 Unity 目錄管線。

範例：使用 spark.sql 存取資料表

您也可以在查詢函式中使用 spark.sql 運算式傳回資料集。 若要從內部資料集讀取，請在資料集名稱前面加上 LIVE.：

@dlt.table
def chicago_customers():
  return spark.sql("SELECT * FROM LIVE.customers_cleaned WHERE city = 'Chicago'")

建立資料表以作為串流作業的目標

使用 create_streaming_table() 函式，透過串流作業建立記錄輸出的目標資料表，包括 apply_changes ()、apply_changes_from_snapshot () 和 @append_flow 輸出記錄。

create_streaming_table(
  name = "<table-name>",
  comment = "<comment>"
  spark_conf={"<key>" : "<value", "<key" : "<value>"},
  table_properties={"<key>" : "<value>", "<key>" : "<value>"},
  partition_cols=["<partition-column>", "<partition-column>"],
  cluster_by = ["<clustering-column>", "<clustering-column>"],
  path="<storage-location-path>",
  schema="schema-definition",
  expect_all = {"<key>" : "<value", "<key" : "<value>"},
  expect_all_or_drop = {"<key>" : "<value", "<key" : "<value>"},
  expect_all_or_fail = {"<key>" : "<value", "<key" : "<value>"},
  row_filter = "row-filter-clause"
)

控制資料表具體化的方式

資料表也提供其具體化的額外控制：

• 指定如何使用 partition_cols 來分割資料表。 您可以使用資料分割來加速查詢。
• 您可以在定義檢視或資料表時設定資料表屬性。 請參閱差異即時資料表資料表屬性。
• 使用 path 設定來設定資料表資料的儲存位置。 如果 path 未設定，資料表資料預設會儲存在管線儲存位置。
• 您可以在結構描述定義中使用產生的資料行。 請參閱範例：指定結構描述和資料分割資料行。

範例：指定結構描述和資料分割資料行

您可以選擇性地使用 Python StructType 或 SQL DDL 字串來指定資料表結構描述。 使用 DDL 字串指定時，定義可以包含產生的資料行。

下列範例會使用 Python StructType 所指定的結構描述建立名為 sales 的資料表：

sales_schema = StructType([
  StructField("customer_id", StringType(), True),
  StructField("customer_name", StringType(), True),
  StructField("number_of_line_items", StringType(), True),
  StructField("order_datetime", StringType(), True),
  StructField("order_number", LongType(), True)]
)

@dlt.table(
  comment="Raw data on sales",
  schema=sales_schema)
def sales():
  return ("...")

下列範例會使用 DDL 字串指定資料表的結構描述、定義產生的資料行，以及定義資料分割資料行：

@dlt.table(
  comment="Raw data on sales",
  schema="""
    customer_id STRING,
    customer_name STRING,
    number_of_line_items STRING,
    order_datetime STRING,
    order_number LONG,
    order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime))
    """,
  partition_cols = ["order_day_of_week"])
def sales():
  return ("...")

根據預設，如果您未指定結構描述，差異即時資料表會從 table 定義推斷結構描述。

設定串流資料表以忽略來源串流資料表中的變更

根據預設，串流資料表需要僅限附加的來源。 當串流資料表使用另一個串流資料表做為來源，而來源串流資料表需要更新或刪除時，例如 GDPR「被遺忘的權利」處理時，skipChangeCommits 可以在讀取來源串流資料表時設定旗標來忽略這些變更。 如需此旗標的詳細資訊，請參閱忽略更新和刪除。

@table
def b():
   return spark.readStream.option("skipChangeCommits", "true").table("LIVE.A")

範例：定義資料表條件約束

指定結構描述時，可以定義主索引鍵和外部索引鍵。 條件約束僅供參考，不會強制執行。 請參閱 SQL 語言參考中的 CONSTRAINT 子句。

下列範例會定義具有主索引鍵和外部索引鍵條件約束的資料表：

@dlt.table(
   schema="""
    customer_id STRING NOT NULL PRIMARY KEY,
    customer_name STRING,
    number_of_line_items STRING,
    order_datetime STRING,
    order_number LONG,
    order_day_of_week STRING GENERATED ALWAYS AS (dayofweek(order_datetime)),
    CONSTRAINT fk_customer_id FOREIGN KEY (customer_id) REFERENCES main.default.customers(customer_id)
    """
def sales():
   return ("...")

範例：定義資料列篩選和資料行遮罩

若要建立具有資料列篩選和資料行遮罩的具體化檢視或串流資料表，請使用 ROW FILTER 子句和 MASK 子句。 下列範例示範如何定義具有資料列篩選和資料行遮罩的具體化檢視和串流資料表：

@dlt.table(
   schema="""
    id int COMMENT 'This is the customer ID',
    name string COMMENT 'This is the customer full name',
    region string,
    ssn string MASK catalog.schema.ssn_mask_fn USING COLUMNS (region)
    """,
  row_filter = "ROW FILTER catalog.schema.us_filter_fn ON (region, name)"
def sales():
   return ("...")

如需有關資料列篩選和資料行遮罩的詳細資訊，請參閱發佈具有資料列篩選和資料行遮罩的資料表。

Python 差異即時資料表屬性

下表描述使用差異即時資料表定義資料表和檢視時，您可以指定的選項和屬性：

在差異即時資料表中使用 Python 從變更摘要擷取異動資料

使用 Python API 中的 apply_changes()函式，以使用差異即時資料表異動資料擷取 (CDC) 功能來處理異動資料摘要 (CDF) 中的來源資料。

apply_changes(
  target = "<target-table>",
  source = "<data-source>",
  keys = ["key1", "key2", "keyN"],
  sequence_by = "<sequence-column>",
  ignore_null_updates = False,
  apply_as_deletes = None,
  apply_as_truncates = None,
  column_list = None,
  except_column_list = None,
  stored_as_scd_type = <type>,
  track_history_column_list = None,
  track_history_except_column_list = None
)

若要深入瞭解使用變更摘要處理 CDC，請參閱套用變更 API：使用差異即時資料表簡化異動資料擷取。 如需使用 apply_changes() 函式的範例，請參閱範例：使用 CDF 來源資料處理的 SCD 類型 1 和 SCD 類型 2。

請參閱套用變更 API：使用差異即時資料表簡化異動資料擷取。

使用差異即時資料表中的 Python 從資料庫快照集變更資料擷取

使用 Python API 中的 apply_changes_from_snapshot() 函式，以使用差異即時資料表異動資料擷取 (CDC) 功能來處理來自資料庫快照集的來源資料。

apply_changes_from_snapshot(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = <type>,
  track_history_column_list = None,
  track_history_except_column_list = None
) -> None

若要深入瞭解使用快照處理 CDC，請參閱套用變更 API：使用差異即時資料表簡化異動資料擷取。 如需使用 apply_changes_from_snapshot() 函式的範例，請參閱定期快照擷取和歷程記錄快照擷取範例。

實作 source 引數

apply_changes_from_snapshot() 函式包含 source 引數。 若要處理歷程記錄快照，source 引數應該是 Python Lambda 函式，其會將兩個值傳回給 apply_changes_from_snapshot() 函式：包含要處理的快照資料和快照版本的 Python DataFrame。

以下是 Lambda 函式的簽名：

lambda Any => Optional[(DataFrame, Any)]

• Lambda 函式的引數是最近處理的快照版本。
• Lambda 函式的傳回值為 None 或兩個值的元組：第一個值是包含要處理之快照的 DataFrame。 元組的第二個值是代表快照邏輯順序的快照版本。

實作和呼叫 Lambda 函式的範例：

def next_snapshot_and_version(latest_snapshot_version):
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

apply_changes_from_snapshot(
  # ...
  source = next_snapshot_and_version,
  # ...
)

每次觸發包含 apply_changes_from_snapshot() 函式的管線時，差異即時資料表執行階段都會執行下列步驟：

1. 執行 next_snapshot_and_version 函式以載入下一個快照 DataFrame 和對應的快照版本。
2. 如果沒有 DataFrame 傳回，則會終止執行，且管線更新會標示為完成。
3. 偵測新快照中的變更，並以累加方式將其套用至目標資料表。
4. 返回步驟 #1 以載入下一個快照及其版本。

限制

差異即時資料表 Python 介面有下列限制：

不支援 pivot() 函式。 Spark 中的 pivot 作業需要積極式載入輸入資料，才能計算輸出的結構描述。 差異即時資料表不支援此功能。