Usar Python com pipelines autônomos

Você pode criar e atualizar exibições materializadas autônomas e tabelas de streaming de um bloco de anotações usando Python. Isso permite que você gerencie pipelines autônomos junto com seus outros fluxos de trabalho de notebook baseados em Python.

Há duas maneiras 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 DataFrame. Consulte Defina tabelas com os decoradores de pipelines.
• Submeta as mesmas instruções SQL que um warehouse SQL do Databricks executa, passando-as para spark.sql(). Isso fornece o suporte completo a SQL para views materializadas autônomas e tabelas de streaming, incluindo instruções REFRESH e agendamentos de atualização. Veja Enviar instruções SQL com spark.sql().

O código-fonte Python para pipelines independentes exige um notebook vinculado à computação geral sem servidor. Você não pode usar Python para criar nem atualizar pipelines autônomos a partir de um warehouse do Databricks SQL, porque um warehouse executa instruções SQL, e não notebooks Python. Para usar um warehouse SQL em vez dele, consulte Usar exibições materializadas autônomas e Usar tabelas de streaming autônomas.

Requirements

Para criar e atualizar pipelines autônomos com Python, você precisa de um notebook conectado à computação geral sem servidor no Databricks Runtime 18.1 ou superior. Para obter a lista completa de requisitos, incluindo disponibilidade e permissões regionais, consulte Notebooks.

Defina tabelas com os decoradores de pipeline

Você pode definir uma visualização materializada independente ou uma tabela de streaming com os mesmos decoradores que você usa em um pipeline do Lakeflow. Cada função decorada define uma tabela. Quando você executa a célula, o Azure Databricks cria a tabela e executa um pipeline sem servidor para preenchê-la. A célula retorna quando a atualização termina.

Defina uma visão materializada

Use @dp.materialized_view em uma função que retorne um DataFrame em lote. O exemplo a seguir cria a visualização daily_booking_revenue materializada a partir da bookings tabela no conjunto de dados de exemplo do Wanderbricks:

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 em streaming, use @dp.table em vez disso.

Defina uma tabela de streaming

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

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 visualização materializada em vez disso. A única exceção é replace_where, que sempre resulta em uma tabela de streaming. O exemplo a seguir mantém a receita diária dos check-ins a partir de 1º de julho de 2025 atualizada, sem recalcular datas anteriores:

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 que correspondem ao predicado e recalcula apenas esse intervalo. Consulte Processamento em lote com fluxos REPLACE WHERE.

Atualizar uma tabela

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

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

@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 usando um decorador, nem agendar atualizações com SCHEDULE ou TRIGGER ON UPDATE. Para atualizar um cronograma, defina a tabela em SQL ou agende o notebook como uma tarefa. Consulte Trabalhos do Lakeflow.

Configure a tabela

Os decoradores aceitam os mesmos parâmetros comuns de conjunto de dados que aceitam dentro de um pipeline, incluindo comment, table_properties, partition_cols, cluster_by, schema, e 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"))
  )

Para a lista de parâmetros, veja materialized_view e tabela.

private=True não é suportado, pois uma tabela privada só pode ser lida por outros conjuntos de dados no mesmo pipeline.

APIs sem suporte

Uma tabela independente é um único conjunto de dados com um único fluxo, então as APIs que descrevem relações entre conjuntos de dados não estão disponíveis. Os seguintes geram 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
• Consulte @dp.replace_flow e o parâmetro replace_using que definem fluxos REPLACE USING. Consulte Substituição parcial de instantâneos com fluxos REPLACE USING.
• dp.create_sink
• Expectativas, como @dp.expect e @dp.expect_or_fail

Para usar esses recursos, crie um pipeline Lakeflow em vez disso. Consulte Desenvolver código de pipeline com Python.

Enviar instruções SQL com spark.sql()

Em um notebook Python, passe os mesmos comandos que você executaria em um Databricks SQL warehouse para spark.sql(). A exibição materializada autônoma e a sintaxe da tabela de streaming são idênticas; somente a maneira como você envia a instrução difere. Assim como em um warehouse, cada instrução CREATE ou REFRESH executa um pipeline sem servidor para processar a operação.

A spark sessão está disponível por padrão em blocos de anotações Azure Databricks, portanto, nenhuma importação é necessária.

Criar uma visão materializada

O exemplo a seguir cria a exibição materializada mv1 da tabela base 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
""")

Para obter detalhes completos CREATE MATERIALIZED VIEW , como atualizações agendadas e disparadas, consulte Criar uma exibição materializada.

Criar uma tabela de streaming

O exemplo a seguir cria a tabela de streaming sales a partir da tabela raw_data:

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

Para obter detalhes completos CREATE STREAMING TABLE , incluindo o carregamento de arquivos com o Carregador Automático e o agendamento, consulte Usar tabelas de streaming autônomas.

Atualizar uma exibição materializada ou uma tabela de streaming

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

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

Na computação geral sem servidor, as atualizações são síncronas. Não há suporte para atualizações assíncronas (a ASYNC palavra-chave). Consulte a computação geral sem servidor.

Parametrizar instruções

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

O exemplo a seguir parametriza o nome de exibição materializado e um valor de filtro:

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 IDENTIFIER cláusula.

Executar outras instruções

Você pode executar qualquer instrução independente de visão materializada ou de tabela de streaming em um notebook Python, passando-a para spark.sql(), incluindo instruções para agendar atualizações, alterar uma tabela ou excluir uma tabela. Para entender como usar exibições materializadas e tabelas de streaming, incluindo sintaxe SQL, consulte Usar exibições materializadas autônomas e usar tabelas de streaming autônomas.

Limitações

Exibições materializadas autônomas e tabelas de streaming criadas na computação geral sem servidor têm limitações adicionais, como nenhum suporte para atualizações assíncronas e nenhuma atribuição de custo por tabela. Para obter a lista completa, consulte a computação geral sem servidor.

Como esses pipelines são executados em computação geral sem servidor, em vez de um data warehouse SQL, eles não herdam tags personalizadas de um data warehouse externo. A propagação de tags do warehouse para system.billing.usage se aplica apenas a visualizações materializadas e tabelas de streaming cujas instruções são executadas em um SQL warehouse. Consulte Atribuir custos ao SQL Warehouse com tags personalizadas.

Tabelas definidas com os decoradores de pipeline apresentam as seguintes limitações adicionais:

• Você não pode atualizá-los com uma REFRESH declaração ou atualizar o cronograma com SCHEDULE ou TRIGGER ON UPDATE. Veja Atualizar uma tabela.
• Expectativas, fluxos adicionais, fluxos de captura de dados de alteração (CDC), sinks e visualizações temporárias não são suportados. Veja APIs não suportadas.
• Não há suporte para private=True.

Recursos adicionais

• Requisitos para pipelines independentes
• Usar visualizações materializadas independentes
• Usar tabelas de streaming autônomas
• Agendar atualizações de dados
• Atualização completa para tabelas de streaming
• Referência da linguagem Python para pipelines do Lakeflow