增量实时表 Python 语言参考

本文提供了有关 Delta Live Tables Python 编程接口的详细信息。

有关 SQL API 的信息，请参阅增量实时表 SQL 语言参考。

有关配置自动加载程序的详细信息，请参阅什么是自动加载程序？。

开始之前

使用 Delta Live Tables Python 接口实现管道时，需要考虑以下重要事项：

• 由于在规划和运行管道更新期间会多次调用 Python table() 和 view() 函数，因此不要在这些函数中包含可能产生副作用的代码（例如，修改数据或发送电子邮件的代码）。 为了避免出现意外行为，定义数据集的 Python 函数应该仅包含定义表或视图所需的代码。
• 若要执行发送电子邮件或与外部监控服务集成等操作，特别是在定义数据集的函数中，请使用事件挂钩。 在定义数据集的函数中实现这些操作将导致出现意外行为。
• Python table 和 view 函数必须返回数据帧。 某些对数据帧进行操作的函数不返回数据帧，因此不应使用。 这些操作包括 collect()、count()、toPandas()、save()、saveAsTable() 等函数。 由于数据帧转换是在解析完整数据流图后执行的，因此使用此类操作可能会产生意想不到的副作用。

导入 dlt Python 模块

增量实时表 Python 函数在 dlt 模块中定义。 利用 Python API 实现的管道必须导入此模块：

import dlt

创建 Delta Live Tables 具体化视图或流式表

在 Python 中，Delta Live Tables 根据定义查询来确定是将数据集更新为具体化视图还是流式表。 @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>)

创建 Delta Live Tables 视图

要在 Python 中定义视图，请应用 @view 装饰器。 与 @table 修饰器一样，可以将 Delta Live Tables 中的视图用于静态或流式处理数据集。 下面是使用 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 参数来分配表或视图名称。 以下示例定义了两个不同的数据集：一个将 JSON 文件作为输入源的 taxi_raw 视图，一个将 taxi_raw 视图作为输入的 filtered_data 表：

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 spark.read.table("LIVE.taxi_raw").where(...)

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

示例：访问在同一管道中定义的数据集

要访问在同一管道中定义的数据集，使用 spark.read.table() 或 spark.readStream.table() 函数，在数据集名称前加上 LIVE 关键字：

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

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

示例：从元存储中注册的表读取

如果要从 Hive 元存储中注册的表读取数据，请在函数参数中忽略LIVE关键字，并选择性地使用数据库名称来限定表名称：

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

若要通过示例来了解如何从 Unity Catalog 表读取数据，请参阅将数据引入 Unity Catalog 管道。

示例：使用 spark.sql 访问数据集

你还可以在查询函数中使用 spark.sql 表达式返回数据集。 若要从内部数据集读取数据，请在数据集名称前追加 LIVE.：

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

创建一个表，用作流式处理操作的目标

使用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"
)

参数
name 类型：str 表名称。 此参数是必需的。
comment 类型：str 表的可选说明。
spark_conf 类型：dict 用于执行此查询的 Spark 配置的可选列表。
table_properties 类型：dict 表的表属性可选列表。
partition_cols 类型：array 包含一列或多列的可选列表，用于对表进行分区。
cluster_by 类型：array （可选）在表上启用 liquid 聚类分析，并定义用作聚类分析键的列。 请参阅对 Delta 表使用 liquid 聚类分析。
path 类型：str 表数据的可选存储位置。 如果未设置，系统默认为管道存储位置。
schema 类型：str 或 StructType 表的可选架构定义。 架构可以定义为 SQL DDL 字符串，或使用 Python 定义 StructType。
expect_all expect_all_or_drop expect_all_or_fail 类型：dict 表的可选数据质量约束。 请参阅多个期望。
row_filter（公共预览版） 类型：str 表的可选行筛选器子句。 请参阅发布具有行筛选器和列掩码的表。

控制表的具体化方式

表还提供对其具体化的额外控制：

• 指定如何使用 partition_cols 对表进行分区。 可以使用分区来加快查询速度。
• 可以在定义视图或表时设置表属性。 请参阅 Delta Live Tables 表属性。
• 使用 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 Delta Live Tables 属性

下表描述了在使用 Delta Live Tables 定义表和视图时可以指定的选项和属性：

@table 或 @view
name 类型：str 表或视图的可选名称。 如果未定义，将使用函数名称作为表名或视图名称。
comment 类型：str 表的可选说明。
spark_conf 类型：dict 用于执行此查询的 Spark 配置的可选列表。
table_properties 类型：dict 表的表属性可选列表。
path 类型：str 表数据的可选存储位置。 如果未设置，系统默认为管道存储位置。
partition_cols 类型：a collection of str 包含一列或多列的可选集合（例如，list），用于对表进行分区。
cluster_by 类型：array （可选）在表上启用 liquid 聚类分析，并定义用作聚类分析键的列。 请参阅对 Delta 表使用 liquid 聚类分析。
schema 类型：str 或 StructType 表的可选架构定义。 架构可以定义为 SQL DDL 字符串，或使用 Python StructType 定义。
temporary 类型：bool 创建表，但不发布表的元数据。 temporary 关键字指示 Delta Live Tables 创建可用于管道但不应在管道外部访问的表。 为了缩短处理时间，临时表会在创建它的管道的生存期内持久保留，而不仅仅是一次更新。 默认值为“False”。
row_filter（公共预览版） 类型：str 表的可选行筛选器子句。 请参阅发布具有行筛选器和列掩码的表。

表或视图定义
def <function-name>() 用于定义数据集的 Python 函数。 如果未设置 name 参数，则使用 <function-name> 作为目标数据集名称。
query 一个 Spark SQL 语句，它返回 Spark Dataset 或 Koalas DataFrame。 使用 dlt.read() 或 spark.read.table() 从同一管道中定义的数据集执行完整读取操作。 要读取外部数据集，请使用 spark.read.table() 函数。 不得使用 dlt.read() 读取外部数据集。 由于 spark.read.table() 可用于读取内部数据集、在当前管道外部定义的数据集，并允许指定用于读取数据的选项，因此 Databricks 建议使用该函数而不是 dlt.read() 函数。 使用 spark.read.table() 函数从同一管道中定义的数据集读取数据时，要在函数参数中的数据集名称前加上 LIVE 关键字。 例如，从名为 customers 的数据集读取数据： spark.read.table("LIVE.customers") 还可以使用 spark.read.table() 函数从元存储中注册的表中读取数据，方法是省略 LIVE 关键字，并选择性地使用数据库名称限定表名称： spark.read.table("sales.customers") 使用 dlt.read_stream() 或 spark.readStream.table() 从同一管道中定义的数据集执行流式读取操作。 要从外部数据集执行流式读取，请使用 spark.readStream.table() 函数。 由于 spark.readStream.table() 可用于读取内部数据集、在当前管道外部定义的数据集，并允许指定用于读取数据的选项，因此 Databricks 建议使用该函数而不是 dlt.read_stream() 函数。 要使用 SQL 语法在 Delta Live Tables table 函数中定义查询，请使用 spark.sql 函数。 请参阅示例：使用 spark.sql 访问数据集。 要使用 Python 在 Delta Live Tables table 函数中定义查询，请使用 PySpark 语法。

预期
@expect("description", "constraint") 声明由以下参数确定的数据质量约束： description。 如果某行违反了预期，则在目标数据集中包含该行。
@expect_or_drop("description", "constraint") 声明由以下参数确定的数据质量约束： description。 如果某行违反了预期，则从目标数据集中删除该行。
@expect_or_fail("description", "constraint") 声明由以下参数确定的数据质量约束： description。 如果某行违反了预期，则立即停止执行。
@expect_all(expectations) 声明一个或多个数据质量约束。 expectations 是一个 Python 字典，其中的键是预期说明，值是预期约束。 如果某行违反了其中一个预期，则在目标数据集中包含该行。
@expect_all_or_drop(expectations) 声明一个或多个数据质量约束。 expectations 是一个 Python 字典，其中的键是预期说明，值是预期约束。 如果某行违反了任何预期，则从目标数据集中删除该行。
@expect_all_or_fail(expectations) 声明一个或多个数据质量约束。 expectations 是一个 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 处理的详细信息，请参阅APPLY CHANGES API：使用增量实时表简化变更数据捕获。 有关使用apply_changes()函数的示例，请参阅示例：使用 CDF 源数据处理 SCD 类型 1 和 SCD 类型 2。

请参阅APPLY CHANGES API：使用增量实时表简化变更数据捕获。

参数
target 类型：str 要更新的表的名称。 可以在执行 apply_changes() 函数之前使用 create_streaming_table() 函数创建目标表。 此参数是必需的。
source 类型：str 包含 CDC 记录的数据源。 此参数是必需的。
keys 类型：list 唯一标识源数据中的行的列或列组合。 这用于标识哪些 CDC 事件适用于目标表中的特定记录。 可以指定以下任一项： - 字符串列表：["userId", "orderId"] - Spark SQL col() 函数列表：[col("userId"), col("orderId"] col() 函数的参数不能包含限定符。 例如，可以使用 col(userId)，但不能使用 col(source.userId)。 此参数是必需的。
sequence_by 类型：str 或 col() 指定源数据中 CDC 事件的逻辑顺序的列名。 增量实时表使用此排序来处理乱序到达的更改事件。 可以指定以下任一项： - 字符串："sequenceNum" - Spark SQL col() 函数：col("sequenceNum") col() 函数的参数不能包含限定符。 例如，可以使用 col(userId)，但不能使用 col(source.userId)。 指定的列必须是可排序的数据类型。 此参数是必需的。
ignore_null_updates 类型：bool 允许引入包含目标列子集的更新。 当 CDC 事件匹配现有行并且ignore_null_updates为True时，具有null的列会在目标中保留其现有值。 这也适用于值为 null 的嵌套列。 当ignore_null_updates为False时，会使用null值覆盖现有值。 此参数是可选的。 默认值为 False。
apply_as_deletes 类型：str 或 expr() 指定何时应将 CDC 事件视为 DELETE 而不是更新插入。 为了处理乱序数据，被删除的行被暂时保留为基础 Delta 表中的无效标记，并在元存储中创建一个视图来筛选掉这些无效标记。 保留间隔可以配置为 pipelines.cdc.tombstoneGCThresholdInSeconds 表属性。 可以指定以下任一项： - 字符串："Operation = 'DELETE'" - Spark SQL expr() 函数：expr("Operation = 'DELETE'") 此参数是可选的。
apply_as_truncates 类型：str 或 expr() 指定何时应将 CDC 事件视为完整表 TRUNCATE。 由于此子句会触发目标表的完全截断，因此应仅将其用于需要此功能的特定用例。 仅 SCD 类型 1 支持 apply_as_truncates 参数。 SCD 类型 2 不支持截断操作。 可以指定以下任一项： - 字符串："Operation = 'TRUNCATE'" - Spark SQL expr() 函数：expr("Operation = 'TRUNCATE'") 此参数是可选的。
column_list except_column_list 类型：list 要包含在目标表中的列的子集。 使用 column_list 指定要包含的列的完整列表。 使用 except_column_list 指定要排除的列。 可以将任一值声明为字符串列表或 Spark SQL col() 函数： - column_list = ["userId", "name", "city"]。 - column_list = [col("userId"), col("name"), col("city")] - except_column_list = ["operation", "sequenceNum"] - except_column_list = [col("operation"), col("sequenceNum") col() 函数的参数不能包含限定符。 例如，可以使用 col(userId)，但不能使用 col(source.userId)。 此参数是可选的。 当没有 column_list 或 except_column_list 参数传递给函数时，默认设置是包含目标表中的所有列。
stored_as_scd_type 类型：str 或 int 将记录存储为 SCD 类型 1 还是 SCD 类型 2。 对于 SCD 类型 1，将其设置为 1；对于 SCD 类型 2，将其设置为 2。 此子句是可选的。 默认值为 SCD 类型 1。
track_history_column_list track_history_except_column_list 类型：list 要在目标表中跟踪其历史记录的输出列子集。 使用 track_history_column_list 指定要跟踪的列的完整列表。 使用 使用 track_history_except_column_list 指定要从跟踪中排除的列。 可以将任一值声明为字符串列表或 Spark SQL col() 函数： - track_history_column_list = ["userId", "name", "city"]。 - track_history_column_list = [col("userId"), col("name"), col("city")] - track_history_except_column_list = ["operation", "sequenceNum"] - track_history_except_column_list = [col("operation"), col("sequenceNum") col() 函数的参数不能包含限定符。 例如，可以使用 col(userId)，但不能使用 col(source.userId)。 此参数是可选的。 默认设置是未将 track_history_column_list 或 track_history_except_column_list 参数传递给函数时包含目标表中的所有列。

在增量实时表中使用 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 处理的详细信息，请参阅APPLY CHANGES API：使用增量实时表简化变更数据捕获。 有关使用apply_changes_from_snapshot()函数的示例，请参阅定期快照引入和历史快照引入示例。

参数
target 类型：str 要更新的表的名称。 可以在运行 apply_changes() 函数之前使用 create_streaming_table() 函数创建目标表。 此参数是必需的。
source 类型：str 或 lambda function 要定期拍摄快照的表或视图的名称或返回要处理的快照 DataFrame 的 Python lambda 函数和快照版本。 请参阅实现源参数。 此参数是必需的。
keys 类型：list 唯一标识源数据中的行的列或列组合。 这用于标识哪些 CDC 事件适用于目标表中的特定记录。 可以指定以下任一项： - 字符串列表：["userId", "orderId"] - Spark SQL col() 函数列表：[col("userId"), col("orderId"] col() 函数的参数不能包含限定符。 例如，可以使用 col(userId)，但不能使用 col(source.userId)。 此参数是必需的。
stored_as_scd_type 类型：str 或 int 将记录存储为 SCD 类型 1 还是 SCD 类型 2。 对于 SCD 类型 1，将其设置为 1；对于 SCD 类型 2，将其设置为 2。 此子句是可选的。 默认值为 SCD 类型 1。
track_history_column_list track_history_except_column_list 类型：list 要在目标表中跟踪其历史记录的输出列子集。 使用 track_history_column_list 指定要跟踪的列的完整列表。 使用 使用 track_history_except_column_list 指定要从跟踪中排除的列。 可以将任一值声明为字符串列表或 Spark SQL col() 函数： - track_history_column_list = ["userId", "name", "city"]。 - track_history_column_list = [col("userId"), col("name"), col("city")] - track_history_except_column_list = ["operation", "sequenceNum"] - track_history_except_column_list = [col("operation"), col("sequenceNum") col() 函数的参数不能包含限定符。 例如，可以使用 col(userId)，但不能使用 col(source.userId)。 此参数是可选的。 默认设置是未将 track_history_column_list 或 track_history_except_column_list 参数传递给函数时包含目标表中的所有列。

实现 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 以加载下一个快照及其版本。

限制

Delta Live Tables Python 接口具有以下限制：

不支持 pivot() 函数。 Spark 中的 pivot 操作需要预先加载输入数据以计算输出架构。 Delta Live Tables 不支持此功能。