表

@table修饰器可用于在管道中定义流式表。

要定义流式表，请将@table应用于对数据源执行流式读取的查询，或使用create_streaming_table() 函数。

注释

在较旧的 dlt 模块中， @table 运算符用于创建流式处理表和具体化视图。 @table模块中的pyspark.pipelines运算符仍然以这种方式工作，但 Databricks 建议使用该@materialized_view运算符创建具体化视图。

Syntax

from pyspark import pipelines as dp

@dp.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_auto = False,
  cluster_by = ["<clustering-column>", "<clustering-column>"],
  schema="schema-definition",
  row_filter = "row-filter-clause",
  private = False)
@dp.expect(...)
def <function-name>():
    return (<query>)

参数

@dp.expect() 是可选的 Lakeflow 管道预期子句。 可以包含多个预期。 请参阅 期望值。

参数 类型 Description
函数 function 必填。 从用户定义的查询返回 Apache Spark 流式处理数据帧的函数。
name str 表名称。 如果未提供，则默认为函数名称。
comment str 表的说明。
spark_conf dict 用于执行此查询的 Spark 配置列表
table_properties dict 表的dict的 。
path str 表数据的存储位置。 如果未设置，请使用包含表的架构的托管存储位置。
partition_cols list 用于对表进行分区的一个或多个列的列表。
cluster_by_auto bool 启用表格上的自动液体聚类功能。 这可以与 cluster_by 并定义要用作初始聚类分析键的列，然后根据工作负荷监视和自动键选择更新。 请参阅 自动液体聚类。
cluster_by list 对表启用动态聚类，并定义要用作聚类键的列。 请参阅对表使用 liquid 聚类分析。
schema str 或 StructType 表的架构定义。 架构可以定义为 SQL DDL 字符串，或使用 Python StructType 定义。 有关 DDL 字符串中支持的列属性，请参阅 的CREATE STREAMING TABLE部分。
private bool 创建表，但不将表发布到元存储。 该表可用于管道，但不能在管道外部访问。 专用表在管道的生存期内保留。
默认值为 False。
专用表以前是使用 temporary 参数创建的。
row_filter str （公共预览版）表的行筛选器子句。 请参阅发布具有行筛选器和列掩码的表。

指定架构是可选的，可以使用 PySpark StructType 或 SQL DDL 来完成。 指定架构时，可以选择包括生成的列、列掩码，以及主键和外键。 请参阅：

例子

from pyspark import pipelines as dp

# Specify a schema
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)]
)
@dp.table(
  comment="Raw data on sales",
  schema=sales_schema)
def sales():
  return ("...")

# Specify a schema with SQL DDL, use a generated column, and set clustering columns
@dp.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))
    """,
  cluster_by = ["order_day_of_week", "customer_id"])
def sales():
  return ("...")

# Specify a schema with an identity column
@dp.table(
  comment="Raw data on sales",
  schema="""
    order_id BIGINT GENERATED ALWAYS AS IDENTITY,
    customer_name STRING,
    order_datetime STRING
    """)
def sales():
  return ("...")

# Use automatic liquid clustering to let Databricks choose the clustering columns
@dp.table(
  comment="Raw data on sales",
  cluster_by_auto=True)
def sales():
  return ("...")

# Specify partition columns
@dp.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 ("...")

# Specify table constraints
@dp.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 ("...")

# Specify a row filter and column mask
@dp.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 ("...")
  • Last updated on