Utilize Python com pipelines autónomos

Pode criar e atualizar vistas materializadas autónomas e tabelas de streaming a partir de um caderno usando Python. Isto permite-lhe gerir pipelines independentes juntamente com os seus outros fluxos de trabalho em blocos de notas baseados em Python.

Há duas maneiras de fazer isso:

• Defina a tabela com os decoradores pyspark.pipelines, @dp.materialized_view e @dp.table. Use isto quando a lógica for mais fácil de expressar como código DataFrame. Consulte Definir tabelas com os decoradores de pipelines.
• Submeta as mesmas instruções SQL que um armazém SQL do Databricks executa, passando-as para spark.sql(). Isto fornece-lhe a vista materializada autónoma completa e a interface SQL da tabela de streaming, incluindo instruções REFRESH e agendamentos de atualização. Veja Enviar comandos SQL com spark.sql().

O código-fonte em Python para pipelines independentes requer um notebook associado à computação geral sem servidor. Não é possível usar Python para criar ou atualizar pipelines autónomos a partir de um armazém SQL do Databricks, porque um armazém executa instruções SQL, não blocos de notas Python. Para utilizar um SQL warehouse em alternativa, consulte Utilizar vistas materializadas autónomas e Utilizar tabelas de streaming autónomas.

Requirements

Para criar e atualizar pipelines independentes com Python, precisa de um notebook associado à computação geral sem servidor no Databricks Runtime 18.1 ou superior. Para a lista completa de requisitos, incluindo disponibilidade regional e permissões, consulte Notebooks.

Defina tabelas com decoradores de cadeias de processamento

Pode definir uma vista materializada autónoma ou uma tabela de streaming com os mesmos decoradores que usa num pipeline do Lakeflow. Cada função decorada define uma tabela. Ao executar a célula, o Azure Databricks cria a tabela e executa um pipeline sem servidor para a povoar. A célula volta a aparecer quando a atualização estiver concluída.

Defina uma visão materializada

Use @dp.materialized_view numa função que devolve um DataFrame por lote. O exemplo seguinte cria a vista materializada daily_booking_revenue a partir da tabela bookings do conjunto de dados de exemplo 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 numa função que devolve um DataFrame em fluxo. O exemplo seguinte cria a tabela de transmissão em fluxo bookings_raw a partir de uma leitura em fluxo 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 devolver um DataFrame por lote, @dp.table cria uma vista materializada em vez disso. A única exceção é replace_where, que resulta sempre numa tabela de streaming. O exemplo seguinte mantém atualizadas as receitas diárias dos check-ins em ou após 1 de julho de 2025, 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 elimina as linhas que correspondem ao predicado e recalcula apenas esse intervalo. Consulte o processamento em lote com fluxos REPLACE WHERE.

Atualizar uma tabela

Para atualizar uma tabela que definiste com um decorador, executa novamente o código que a define, por exemplo, reexecutando a célula do caderno, executando o caderno inteiro ou executando o caderno como um trabalho. Cada execução cria a tabela se esta não existir e atualiza-a se existir.

Para reprocessar todos os dados disponíveis na origem, transmita full_refresh=True em qualquer um dos dois 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")

Não pode usar uma instrução REFRESH numa tabela definida por um decorador, nem agendar atualizações com SCHEDULE ou TRIGGER ON UPDATE. Para atualizar um cronograma, define a tabela em SQL ou agenda o caderno como uma tarefa. Veja Lakeflow Jobs.

Configurar a tabela

Os decoradores aceitam os mesmos parâmetros comuns do conjunto de dados que aceitam num 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, porque uma tabela privada só pode ser lida por outros conjuntos de dados no mesmo pipeline.

APIs não suportadas

Uma tabela autónoma é um único conjunto de dados com um único fluxo, por isso 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
• @dp.replace_flow e o parâmetro replace_using, que definem fluxos de REPLACE USING. Veja Substituição parcial de instantâneo com fluxos REPLACE USING.
• dp.create_sink
• Expectativas, como @dp.expect e @dp.expect_or_fail

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

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

Num caderno Python, passe as mesmas instruções que executaria de um armazém SQL do Databricks para spark.sql(). A sintaxe da visualização materializada autónoma e da tabela de streaming é idêntica; Apenas a forma como submete a declaração é diferente. Tal como acontece num armazém, cada instrução CREATE ou REFRESH executa um pipeline sem servidor para processar a operação.

A sessão spark está disponível por predefinição nos notebooks do Azure Databricks, pelo que não é necessário importar nada.

Criar uma visão materializada

O exemplo a seguir cria a exibição mv1 materializada a partir da tabela base_table1base:

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 todos os detalhes CREATE MATERIALIZED VIEW, como atualizações agendadas e desencadeadas, consulte Criar uma vista materializada.

Criar uma tabela de transmissão

O exemplo seguinte 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 todos os detalhes, incluindo o carregamento de ficheiros com Auto Loader e o agendamento, consulte CREATE STREAMING TABLE.

Atualize uma visualização materializada ou uma tabela de streaming

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

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

Em computação geral sem servidor, as atualizações são síncronas. As atualizações assíncronas (a ASYNC palavra-chave) não são suportadas. Ver Computação geral sem servidor.

Parametrizar sentenças

Para passar valores do seu código Python para uma instrução em vez de os codificar fixamente, use marcadores de parâmetros nomeados no SQL e forneça os seus valores através do args argumento de spark.sql(). Use um marcador, como :min_sales diretamente, para valores literais. Envolva o marcador IDENTIFIER() apenas quando o parâmetro for um nome de objeto, como uma tabela, vista ou esquema, porque identificadores não podem ser substituídos por valores simples de cadeia.

O exemplo seguinte parametriza tanto o nome da visualização materializada como 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 mais informações, veja Marcadores de parâmetros e IDENTIFIER cláusula.

Executar outras declarações

Pode executar qualquer instrução de visualização materializada autónoma ou de tabela em streaming a partir de um caderno Python, passando-a para spark.sql(), incluindo instruções para agendar atualizações, alterar uma tabela ou eliminar uma tabela. Para compreender como usar vistas materializadas e tabelas de streaming, incluindo sintaxe SQL, veja Usar vistas materializadas autónomas e Usar tabelas de streaming independentes.

Limitations

Vistas materializadas autónomas e tabelas de streaming criadas em computação geral sem servidor têm limitações adicionais, como a ausência de suporte para atualizações assíncronas e a ausência de atribuição de custos por tabela. Para a lista completa, veja Computação geral sem servidor.

Como estes pipelines são executados em computação geral sem servidor, em vez de num armazém de dados SQL, não herdam etiquetas personalizadas de um armazém que os engloba. A propagação da etiqueta do warehouse aplica-se apenas a system.billing.usage visualizações materializadas e tabelas de streaming cujas instruções são executadas num SQL warehouse. Veja Custos de atributos para o SQL warehouse com tags personalizadas.

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

• Não pode atualizá-los com uma REFRESH declaração ou atualizar o calendário com SCHEDULE ou TRIGGER ON UPDATE. Veja Atualizar uma tabela.
• Expectativas, fluxos adicionais, fluxos de captura de dados de alteração (CDC), sumidouros e vistas temporárias não são suportados. Veja APIs não suportadas.
• private=True não é suportado.

Recursos adicionais

• Requisitos para pipelines independentes
• Use vistas materializadas independentes
• Utilize tabelas de streaming autónomas
• Atualizações de horários
• Atualização completa para tabelas de streaming
• Referência da linguagem Python para pipelines do Lakeflow