mesa

O decorador @table pode ser usado para definir tabelas de transmissão em um Lakeflow pipeline ou em um pipeline independente.

Para definir uma tabela de transmissão, aplique @table a uma consulta que executa uma leitura de transmissão em uma fonte de dados ou use a função create_streaming_table().

nota

No módulo dlt mais antigo, o operador @table era usado para criar tabelas de transmissão e visualizações materializadas. O operador @table no módulo pyspark.pipelines ainda funciona dessa maneira, mas Databricks recomenda usar o operador @materialized_view para criar uma visualização materializada.

Você também pode aplicar @table fora de um pipeline, em um notebook em compute geral serverless, para definir uma tabela de transmissão autônoma, com algumas limitações. Consulte Definir tabelas com os decoradores de pipelines.

Sintaxe

Python

from pyspark import pipelines as dp

@dp.table(
  name="<name>",
  comment="<comment>",
  spark_conf={&quot;&lt;key&gt;&quot; : &quot;&lt;value&gt;&quot;, &quot;&lt;key&gt;&quot; : &quot;&lt;value&gt;&quot;},
  table_properties={&quot;&lt;key&gt;&quot; : &quot;&lt;value&gt;&quot;, &quot;&lt;key&gt;&quot; : &quot;&lt;value&gt;&quot;},
  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,
  replace_using = ["<key-column>", "<key-column>"],
  sequence_by = "<sequence-column>")
@dp.expect(...)
def <function-name>():
    return (<query>)

Parâmetros

@dp.expect() é uma cláusula de expectativa opcional do LakeFlow Pipelines. É possível incluir múltiplas expectativas. Consulte Expectativas.

Parâmetro	Tipo	Descrição
função	function	Obrigatório. Uma função que retorna um DataFrame de streaming Apache Spark streaming de uma consulta definida pelo usuário.
name	str	O nome da tabela. Se não for fornecido, o padrão será o nome da função.
comment	str	Uma descrição para a tabela.
spark_conf	dict	Uma lista de configurações do Spark para a execução desta consulta
table_properties	dict	Um dict de propriedades de tabela para a tabela.
path	str	Um local de armazenamento para dados da tabela. Se não for definido, use o local de armazenamento gerenciar para o esquema que contém a tabela.
partition_cols	list	Uma lista de uma ou mais colunas a serem usadas para particionar a tabela.
cluster_by_auto	bool	Ativar o liquid clustering automático na tabela. Isso pode ser combinado com cluster_by e definir as colunas a serem usadas como chaves de agrupamento iniciais, seguido por monitoramento e atualizações automáticas de seleção de chaves com base na carga de trabalho. Consulte Clustering Líquido Automático.
cluster_by	list	Habilite clustering líquido na tabela e defina as colunas a serem usadas como chave clustering . Consulte Usar clustering líquido para tabelas.
schema	str ou StructType	Uma definição de esquema para a tabela. Esquemas podem ser definidos como uma string SQL DDL ou com um Python StructType. Para as propriedades de coluna suportadas na string DDL, consulte a seção Parâmetros de CREATE STREAMING TABLE.
private	bool	Crie uma tabela, mas não publique a tabela no metastore. Essa tabela está disponível para o pipeline, mas não está acessível fora do pipeline. Tabelas privadas persistem durante a vida útil do pipeline. O padrão é False. Tabelas privadas foram criadas anteriormente com o parâmetro temporary .
row_filter	str	(Visualização pública) Uma cláusula de filtro de linha para a tabela. Consulte Publicar tabelas com filtros de linha e máscaras de coluna.
replace_using	list	(Beta) As colunas key que identificam quais linhas de destino substituir, o que define a query da tabela como um fluxo REPLACE USING. Especifique pelo menos uma coluna. Requer uma fonte de transmissão e sequence_by. Consulte Substituição parcial de Snapshot com fluxos REPLACE USING.
sequence_by	str ou Column	(Beta) A coluna que ordena as atualizações para um fluxo replace_using, de modo que a sequência mais alta para uma key vença. Necessário sempre que replace_using estiver definido.

Parâmetro	Tipo	Descrição
função	function	Obrigatório. Uma função que retorna um DataFrame de streaming Apache Spark streaming de uma consulta definida pelo usuário.
name	str	O nome da tabela. Se não for fornecido, o padrão será o nome da função.
comment	str	Uma descrição para a tabela.
spark_conf	dict	Uma lista de configurações do Spark para a execução desta consulta
table_properties	dict	Um dict de propriedades de tabela para a tabela.
path	str	Um local de armazenamento para dados da tabela. Se não for definido, use o local de armazenamento gerenciar para o esquema que contém a tabela.
partition_cols	list	Uma lista de uma ou mais colunas a serem usadas para particionar a tabela.
cluster_by_auto	bool	Ativar o liquid clustering automático na tabela. Isso pode ser combinado com cluster_by e definir as colunas a serem usadas como chaves de agrupamento iniciais, seguido por monitoramento e atualizações automáticas de seleção de chaves com base na carga de trabalho. Consulte Clustering Líquido Automático.
cluster_by	list	Habilite clustering líquido na tabela e defina as colunas a serem usadas como chave clustering . Consulte Usar clustering líquido para tabelas.
schema	str ou StructType	Uma definição de esquema para a tabela. Esquemas podem ser definidos como uma string SQL DDL ou com um Python StructType. Para as propriedades de coluna suportadas na string DDL, consulte a seção Parâmetros de CREATE STREAMING TABLE.
private	bool	Crie uma tabela, mas não publique a tabela no metastore. Essa tabela está disponível para o pipeline, mas não está acessível fora do pipeline. Tabelas privadas persistem durante a vida útil do pipeline. O padrão é False. Tabelas privadas foram criadas anteriormente com o parâmetro temporary .
row_filter	str	(Visualização pública) Uma cláusula de filtro de linha para a tabela. Consulte Publicar tabelas com filtros de linha e máscaras de coluna.
replace_using	list	(Beta) As colunas key que identificam quais linhas de destino substituir, o que define a query da tabela como um fluxo REPLACE USING. Especifique pelo menos uma coluna. Requer uma fonte de transmissão e sequence_by. Consulte Substituição parcial de Snapshot com fluxos REPLACE USING.
sequence_by	str ou Column	(Beta) A coluna que ordena as atualizações para um fluxo replace_using, de modo que a sequência mais alta para uma key vença. Necessário sempre que replace_using estiver definido.

Especificar um esquema é opcional e pode ser feito com PySpark StructType ou SQL DDL. Ao especificar um esquema, você pode incluir opcionalmente colunas geradas, máscaras de coluna e chaves primária e estrangeira. Ver:

• Colunas geradas pelo Delta Lake
• Restrições no Databricks
• Publique tabelas com filtros de linha e máscaras de coluna.

Exemplos

Python

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 ("...")

# Define the table's query as a REPLACE USING flow, keeping the latest row for each payment_id
@dp.table(
  name="payments_current",
  replace_using=["payment_id"],
  sequence_by="payment_date")
def payments_current():
  return spark.readStream.table("samples.wanderbricks.payments")