Pular para o conteúdo principal
Última atualização em

Usar Python com pipelines autônomos

info

Beta

A criação e o refresh de views materializadas e tabelas de transmissão autônomas a partir de um Notebook em compute geral Serverless está em Beta e disponível em regiões selecionadas. Consulte Notebooks.

Você pode criar e refresh view materializadas autônomas e tabelas de transmissão de um Notebook usando Python. Isso permite que você gerencie pipelines autônomos juntamente com seus outros fluxos de trabalho de notebook baseados em Python.

Há duas formas de fazer isso:

  • Defina a tabela com os decoradores pyspark.pipelines, @dp.materialized_view e @dp.table. Use isso quando a lógica for mais fácil de expressar como código de DataFrame. Consulte Definir tabelas com os decoradores de pipelines.
  • Envie as mesmas instruções SQL que um warehouse do Databricks SQL executa passando-as para spark.sql(). Isso dá a você a superfície SQL completa e autônoma de view materializada e tabela de transmissão, incluindo declarações REFRESH e programar de refresh. Consulte Enviar instruções SQL com spark.sql().

A fonte Python para pipelines autônomos requer um notebook anexado ao compute geral serverless . Você não pode usar Python para criar ou refresh pipelines autônomos de um Databricks SQL warehouse, pois um warehouse executa comandos SQL, não Notebooks Python. Para usar um SQL Warehouse em vez disso, consulte Usar views materializadas autônomas e Usar tabelas de transmissão autônomas.

Requisitos​

Para criar e refresh pipelines autônomos com Python, você precisa de um Notebook anexado ao compute geral Serverless no Databricks Runtime 18.1 ou acima. Para ver a lista completa de requisitos, incluindo disponibilidade regional e permissões, consulte Notebooks.

Definir tabelas com os decoradores de pipelines​

Você pode definir uma view materializada autônoma ou tabela de transmissão com os mesmos decoradores usados em um pipeline Lakeflow. Cada função decorada define uma tabela. Quando você executa a célula, o Databricks cria a tabela e executa um pipeline serverless para preenchê-la. A célula retorna quando a atualização termina.

atenção

Os decoradores de pipelines requerem a versão do ambiente serverless 5 ou acima.

Definir uma view materializada​

Use @dp.materialized_view em uma função que retorna um DataFrame de lote. O exemplo a seguir cria a view materializada daily_booking_revenue a partir da tabela bookings no dataset de exemplo Wanderbricks:

Python
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"))
  )

Para definir uma tabela a partir de uma leitura de transmissão, use @dp.table em seu lugar.

Definir uma tabela de transmissão​

Use @dp.table em uma função que retorna um DataFrame de transmissão. O exemplo a seguir cria a tabela de transmissão bookings_raw a partir de uma leitura de transmissão da mesma tabela bookings:

Python
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")

Se a função retornar um DataFrame em lote, @dp.table criará uma view materializada. A única exceção é replace_where, que sempre resulta em uma tabela de transmissão. O exemplo a seguir mantém a receita diária de check-ins em ou após 1º de julho de 2025 atualizada, sem recalcular datas anteriores:

Python
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"))
  )

Cada execução exclui as linhas correspondentes ao predicado e recalcula apenas esse intervalo. Consulte Processamento em lote com fluxos REPLACE WHERE.

Refresh de uma tabela​

Para refresh uma tabela que você definiu com um decorador, execute o código que a define novamente, por exemplo, executando novamente a célula do notebook, executando o notebook inteiro ou executando o notebook como um Job. Cada execução cria a tabela se ela não existir e a refresh se existir.

Para reprocessar todos os dados disponíveis na fonte, passe full_refresh=True em qualquer decorador:

Python
@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")

Você não pode usar uma instrução REFRESH em uma tabela definida com um decorador ou programar atualizações com SCHEDULE ou TRIGGER ON UPDATE. Para refresh em programar, defina a tabela em SQL ou programe o Notebook como um Job. Consulte Lakeflow Jobs.

Configurar a tabela​

Os decoradores aceitam os mesmos parâmetros comuns de dataset aceitos dentro de um pipeline, incluindo comment, table_properties, partition_cols, cluster_by, schema e spark_conf:

Python
@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"))
  )

Para obter a lista de parâmetros, consulte materialized_view e table.

private=True não é compatível, pois uma tabela privada só pode ser lida por outros datasets no mesmo pipeline.

APIs sem compatibilidade​

Uma tabela autônoma é um dataset único com um fluxo único, portanto, as APIs que descrevem as relações entre os datasets não estão disponíveis. O seguinte gera um erro fora de um pipeline:

  • @dp.temporary_view e dp.create_streaming_table
  • @dp.append_flow e outros fluxos adicionais
  • dp.create_auto_cdc_flow e dp.create_auto_cdc_from_snapshot_flow
  • @dp.replace_flow e o parâmetro replace_using, que definem fluxos REPLACE USING. Consulte Substituição parcial de snapshot com fluxos REPLACE USING.
  • dp.create_sink
  • Expectativas, como @dp.expect e @dp.expect_or_fail

Para usá-los, crie um LakeFlow Pipelines em vez disso. Consulte Desenvolver código de pipeline com Python.

Enviar comandos SQL com spark.sql()​

Em um Notebook Python, passe as mesmas instruções que você faria a execução em um Databricks SQL warehouse para spark.sql(). A sintaxe de view materializada independente e tabela de transmissão é idêntica; apenas a maneira como você envia a declaração é diferente. Assim como em um warehouse, cada declaração CREATE ou REFRESH executa um pipeline serverless para processar a operação.

A sessão spark está disponível por default em Databricks Notebooks, portanto, não é necessária importação.

Criar uma view materializada​

O exemplo a seguir cria a view materializada mv1 a partir da tabela-base base_table1:

Python
spark.sql("""
  CREATE OR REPLACE MATERIALIZED VIEW mv1
  AS SELECT
    date,
    sum(sales) AS sum_of_sales
  FROM base_table1
  GROUP BY date
""")

Para obter detalhes completos sobre CREATE MATERIALIZED VIEW, como refresh agendadas e acionadas por Trigger, consulte Criar uma view materializada.

Criar uma tabela de transmissão​

O exemplo a seguir cria a tabela de transmissão sales a partir da tabela raw_data:

Python
spark.sql("""
  CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT product, price FROM STREAM raw_data
""")

Para obter detalhes completos sobre CREATE STREAMING TABLE, incluindo carregamento de arquivos com Auto Loader e programação, consulte Usar tabelas de transmissão autônomas.

Refresh uma view materializada ou tabela de transmissão​

Use uma declaração REFRESH para atualizar uma tabela autônoma com os dados mais recentes de sua fonte:

Python
spark.sql("REFRESH MATERIALIZED VIEW mv1")
spark.sql("REFRESH STREAMING TABLE sales")

No compute geral Serverless, os refresh são síncronos. Não há suporte para refresh assíncronas (palavra-chave ASYNC). Consulte Compute geral serverless.

Parametrizar instruções​

Para passar valores do seu código Python para uma instrução em vez de codificá-los, use marcadores de parâmetros nomeados no SQL e forneça seus valores por meio do argumento args de spark.sql(). Use um marcador como :min_sales diretamente para valores literais. Coloque o marcador em IDENTIFIER() somente quando o parâmetro for um nome de objeto, como uma tabela, view ou esquema, porque os identificadores não podem ser substituídos como valores de string simples.

O exemplo a seguir parametrizar tanto o nome da view materializada quanto um valor de filtro:

Python
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,
})

Para obter mais informações, consulte Marcadores de parâmetro e Cláusula IDENTIFIER.

Execução outros comandos​

Você pode fazer a execução de qualquer instrução de view materializada autônoma ou tabela de transmissão a partir de um Notebook Python passando-a para spark.sql(), incluindo instruções para programar refresh, alterar uma tabela ou remover uma tabela. Para entender como usar views materializadas e tabelas de transmissão, incluindo a sintaxe SQL, consulte Usar views materializadas autônomas e Usar tabelas de transmissão autônomas.

Limitações​

Views materializadas e tabelas de transmissão autônomas criadas em compute geral Serverless têm limitações adicionais, como falta de suporte para refresh e sem atribuição de custo por tabela. Para obter a lista completa, consulte Compute geral serverless.

Como esses pipelines estão em execução em Serverless compute geral em vez de um SQL Warehouse, eles não herdam tags personalizadas de um warehouse delimitador. A propagação de tags do warehouse para system.billing.usage aplica-se apenas a materialized views e transmissão tables cujas instruções têm execução a partir de um SQL Warehouse. Consulte Atribuir custos ao SQL warehouse com tags personalizadas.

As tabelas definidas com os decoradores de pipelines têm as seguintes limitações adicionais:

  • Não é possível fazer refresh delas com uma instrução REFRESH ou programar refreshes com SCHEDULE ou TRIGGER ON UPDATE. Veja Refresh de uma tabela.
  • Expectativas, fluxos adicionais, fluxos de captura de dados de alterações (CDC), coletores e view temporárias não são compatíveis. Consulte APIs não suportadas.
  • private=True não é suportado.

Outros recursos​

Nesta página