可以在笔记本中使用 Python 创建和刷新独立的物化视图和流式表。 这样就可以与其他基于Python的笔记本工作流一起管理独立管道。
可通过两种方式实现此目的：
- 定义带有
pyspark.pipelines装饰器的表格，@dp.materialized_view和@dp.table。 当逻辑更容易用DataFrame代码表达时，就用这个方法。 参见 “定义带有管道装饰器的表格”。 - 将 Databricks SQL 仓库运行的相同 SQL 语句传递给
spark.sql()以进行提交。 这为你提供完整的独立物化视图和流式表 SQL 接口，包括REFRESH语句和刷新计划。 请参阅使用spark.sql()提交 SQL 语句。
独立管道的 Python 源代码需要连接到无服务器通用计算的笔记本。 不能使用Python从 Databricks SQL 仓库创建或刷新独立管道，因为仓库运行 SQL 语句，而不是Python笔记本。 若要改用 SQL 仓库，请参阅 “使用独立具体化视图 ”和 “使用独立流式处理表”。
Requirements
若要使用 Python 创建和刷新独立管道，您需要一个连接到基于 Databricks Runtime 18.1 或更高版本的无服务器通用计算的笔记本。 有关要求的完整列表，包括区域可用性和权限，请参阅 Notebooks。
定义带有管道装饰器的表格
你可以使用与在 Lakeflow 管道中相同的装饰器来定义独立的物化视图或流式表。 每个装饰函数定义一个表。 当你运行单元时，Azure Databricks会生成表，并运行无服务器流水线来填充它。 更新完成后，单元格会返回。
Warning
管道装饰器需要 无服务器环境版本 5 或更高版本。
定义一个具体化的视图
将 @dp.materialized_view 用于返回批量 DataFrame 的函数。 以下示例从daily_booking_revenue中的bookings表创建物化视图：
from pyspark import pipelines as dp
from pyspark.sql import functions as F
@dp.materialized_view(name="main.default.daily_booking_revenue")
def daily_booking_revenue():
return (
spark.read.table("samples.wanderbricks.bookings")
.groupBy("check_in")
.agg(F.sum("total_amount").alias("total_revenue"))
)
要从流式读取中定义表，请用 @dp.table 代替。
定义流式表
对返回流式 DataFrame 的函数使用 @dp.table。 以下示例是通过对同一bookings表的流式读取创建流式表bookings_raw：
from pyspark import pipelines as dp
@dp.table(name="main.default.bookings_raw")
def bookings_raw():
return spark.readStream.table("samples.wanderbricks.bookings").select("booking_id", "property_id", "total_amount")
如果函数返回的是批处理数据帧，则 @dp.table 会创建一个具体化视图。 唯一的例外是 replace_where，它总是生成一个流式表。 以下示例可使 2025 年 7 月 1 日及之后签到的每日收入保持最新，而无需重新计算更早日期的数据：
from pyspark import pipelines as dp
from pyspark.sql import functions as F
@dp.table(
name="main.default.booking_revenue_rw",
replace_where=F.col("check_in") >= F.to_date(F.lit("2025-07-01")),
)
def booking_revenue_rw():
return (
spark.read.table("samples.wanderbricks.bookings")
.groupBy("check_in")
.agg(F.sum("total_amount").alias("total_revenue"))
)
每次运行都会删除与谓词匹配的行，并只重新计算该区间。 请参阅使用 REPLACE WHERE 流的批处理。
刷新表
要刷新你用装饰器定义的表格，可以重新运行定义它的代码，比如重运行笔记本单元、整个笔记本，或者把笔记本当作作业运行。 每次运行如果不存在该表，都会创建该表，如果存在则刷新。
要重新处理源中所有可用数据，请在两个装饰器中的任意一个上都传递 full_refresh=True：
@dp.table(name="main.default.bookings_raw", full_refresh=True)
def bookings_raw():
return spark.readStream.table("samples.wanderbricks.bookings").select("booking_id", "property_id", "total_amount")
你不能在使用装饰器定义的表上使用 REFRESH 语句，也不能使用 SCHEDULE 或 TRIGGER ON UPDATE 来安排刷新计划。 要刷新调度，可以在SQL中定义表，或者将笔记本调度为作业。 请参阅 Lakeflow Jobs。
配置表格
装饰器可接受与其在管道内可接受的通用数据集参数相同的参数，包括 comment、table_properties、partition_cols、schema、cluster_by 和 spark_conf：
@dp.materialized_view(
name="main.default.daily_booking_revenue",
comment="Daily booking revenue.",
table_properties={"quality": "gold"},
cluster_by=["check_in"],
)
def daily_booking_revenue():
return (
spark.read.table("samples.wanderbricks.bookings")
.groupBy("check_in")
.agg(F.sum("total_amount").alias("total_revenue"))
)
参数列表请参见 materialized_view 和 表格。
private=True 不支持，因为私有表只能被同一流水线内的其他数据集读取。
不支持的 API
独立表是单一数据集和单一流程，因此无法使用描述数据集之间关系的API。 以下提示会在管道外引发错误：
-
@dp.temporary_view和dp.create_streaming_table -
@dp.append_flow以及其他额外的流量 -
dp.create_auto_cdc_flow和dp.create_auto_cdc_from_snapshot_flow -
@dp.replace_flow以及replace_using参数，它们定义了 REPLACE USING 流程。 参见 使用 REPLACE USING 流程进行部分快照替换。 dp.create_sink- 期望，例如
@dp.expect和@dp.expect_or_fail
要使用这些，请改为创建一个 Lakeflow 管道。 请参阅 使用 Python 开发管道代码。
使用 spark.sql() 提交 SQL 语句
在Python笔记本中，将从 Databricks SQL 仓库运行的相同语句传递给 spark.sql()。 独立物化视图和流式表的语法完全相同；区别仅在于提交语句的方式。 与仓库一样，每个 CREATE 或 REFRESH 语句都运行无服务器管道来处理操作。
默认情况下，会话spark在Azure Databricks笔记本中可用，因此无需导入。
创建具体化视图
以下示例从基表 mv1 创建具体化视图 base_table1：
spark.sql("""
CREATE OR REPLACE MATERIALIZED VIEW mv1
AS SELECT
date,
sum(sales) AS sum_of_sales
FROM base_table1
GROUP BY date
""")
有关完整 CREATE MATERIALIZED VIEW 详细信息（例如计划和触发式刷新），请参阅创建具体化视图。
创建流表
以下示例从sales表创建流式表raw_data：
spark.sql("""
CREATE OR REFRESH STREAMING TABLE sales
AS SELECT product, price FROM STREAM raw_data
""")
有关完整 CREATE STREAMING TABLE 详细信息（包括使用自动加载器加载文件和计划任务），请参阅使用独立流式表。
刷新具体化视图或流式表
使用 REFRESH 语句，利用其数据源中的最新数据更新独立表：
spark.sql("REFRESH MATERIALIZED VIEW mv1")
spark.sql("REFRESH STREAMING TABLE sales")
在无服务器常规计算中，刷新是同步的。 不支持异步刷新（ ASYNC 关键字）。 请参阅 无服务器常规计算。
参数化语句
若要将Python代码中的值传递到语句而不是硬编码，请在 SQL 中使用命名参数标记，并通过args参数spark.sql()提供其值。 对于字面值，请直接使用诸如 :min_sales 之类的标记。 仅当参数是对象名称（如表、视图或架构）时，才会包装标记 IDENTIFIER() ，因为标识符不能替换为纯字符串值。
以下示例将物化视图名称和筛选值都参数化：
mv_name = "main.sales.regional_sales"
min_sales = 1000
spark.sql("""
CREATE OR REPLACE MATERIALIZED VIEW IDENTIFIER(:mv)
AS SELECT
region,
sum(sales) AS sum_of_sales
FROM base_table1
WHERE sales > :min_sales
GROUP BY region
""", args={
"mv": mv_name,
"min_sales": min_sales,
})
有关详细信息，请参阅 参数标记 和 IDENTIFIER 子句。
运行其他语句
您可以在 Python 笔记本中将任何独立的物化视图或流式表语句传递给 spark.sql() 来运行，包括用于安排刷新、修改表或删除表的语句。 若要了解如何使用具体化视图和流式处理表（包括 SQL 语法），请参阅使用独立具体化视图和使用独立流式处理表。
Limitations
在无服务器常规计算上创建的独立具体化视图和流式处理表具有额外的限制，例如不支持异步刷新，也不支持每表成本归因。 有关完整列表，请参阅 无服务器常规计算。
由于这些流水线运行在无服务器的通用计算上，而非SQL仓库，因此它们不会继承来自包围仓库的自定义标签。 仓库标记传播到 system.billing.usage 仅适用于其语句从 SQL 仓库运行的具体化视图和流式处理表。 请参阅 使用自定义标记的 SQL 仓库的属性成本。
管道装饰器定义的表格还有以下额外限制：
- 你不能使用
REFRESH语句刷新它们，也不能使用SCHEDULE或TRIGGER ON UPDATE来计划刷新。 参见 “刷新表格”。 - 不支持预期规则、其他流、更改数据捕获 (CDC) 流、接收器和临时视图。 参见 不支持的API。
- 不支持
private=True。