如何在Notebook中测试Delta Live Table代码？无需创建DLT管道的简便方法

1. 启用DLT交互式开发模式 #

创建带有特定Spark配置的集群，即可在Notebook中直接运行DLT代码片段：

• 新建集群时，在Spark配置中添加：spark.databricks.dtlt.enableInteractiveMode true
• 将Notebook挂载到该集群后，就能直接执行带@dlt.table()装饰器的代码，调用函数生成DataFrame查看结果，无需创建正式DLT管道。

2. 拆分逻辑为可独立测试的函数 #

把DLT表的转换逻辑拆成单独函数，先在Notebook里验证函数逻辑，再集成到DLT代码中：

# 定义独立的转换函数
def filter_valid_orders(raw_orders_df):
    return raw_orders_df.filter(raw_orders_df.status.isin("completed", "shipped"))

# 在Notebook中测试函数
test_data = spark.createDataFrame([
    (1, "completed", 100),
    (2, "cancelled", 50),
    (3, "shipped", 75)
], ["order_id", "status", "amount"])
filtered_df = filter_valid_orders(test_data)
filtered_df.show()  # 验证结果是否符合预期

# 集成到DLT表
import dlt
@dlt.table(comment="Validated order data")
def valid_orders():
    raw_df = dlt.read("raw_orders")
    return filter_valid_orders(raw_df)

3. 使用dlt.test()编写单元测试 #

DLT提供dlt.test()方法，可在Notebook中直接验证转换逻辑的正确性：

import dlt

@dlt.table
def raw_orders():
    return spark.read.csv("/path/to/raw/orders", header=True)

def filter_valid_orders(raw_df):
    return raw_df.filter(raw_df.status.isin("completed", "shipped"))

# 编写测试用例
dlt.test(
    name="test_valid_orders_filter",
    test_df=spark.createDataFrame([
        (1, "completed"), (2, "cancelled"), (3, "shipped")
    ], ["order_id", "status"]),
    expect=spark.createDataFrame([
        (1, "completed"), (3, "shipped")
    ], ["order_id", "status"]),
    func=filter_valid_orders
)

运行代码会直接输出测试结果，确认转换逻辑是否达标。

4. 直接调用DLT表函数查看结果 #

在启用交互式模式的集群中，定义完@dlt.table()装饰的函数后，可直接调用函数生成DataFrame，查看数据是否符合预期：

import dlt

@dlt.table
def cleaned_users():
    return spark.read.json("/path/to/raw/users").filter("age >= 18").dropDuplicates(["user_id"])

# 直接调用函数查看结果
cleaned_users().show(10)

内容的提问来源于stack exchange，提问作者Rajib Deb