Uso de Python con canalizaciones independientes

Importante

La creación y actualización de vistas materializadas independientes y tablas de streaming desde un cuaderno en recursos de proceso general sin servidor se encuentra en fase beta y está disponible en determinadas regiones. Consulte Portátiles.

Puede crear y actualizar vistas materializadas independientes y tablas de streaming desde un cuaderno mediante Python. Esto le permite administrar canalizaciones independientes junto con los demás flujos de trabajo de cuaderno basados en Python.

Hay dos maneras de hacerlo:

  • Defina la tabla con los pyspark.pipelines decoradores, @dp.materialized_view y @dp.table. Úsalo cuando la lógica sea más fácil de expresar como código DataFrame. Consulte Definir tablas con los decoradores de canalizaciones.
  • Envía las mismas sentencias SQL que ejecuta un almacén SQL de Databricks pasándolas a spark.sql(). Esto te proporciona la vista materializada independiente completa y la superficie SQL de la tabla de streaming, incluidas las instrucciones REFRESH y las programaciones de actualización. Consulte Enviar sentencias SQL con spark.sql().

El código fuente de Python para pipelines independientes requiere un cuaderno vinculado a entorno de proceso general sin servidor. No se puede usar Python para crear o actualizar pipelines independientes desde un almacén SQL de Databricks, porque un almacén SQL ejecuta sentencias SQL, no cuadernos de Python. Para usar una instancia de SQL Warehouse en su lugar, consulte Uso de vistas materializadas independientes y Uso de tablas de streaming independientes.

Requirements

Para crear y renovar canalizaciones independientes con Python, se necesita un notebook conectado a un entorno de cómputo general sin servidor en Databricks Runtime 18.1 o superior. Para obtener la lista completa de requisitos, incluida la disponibilidad regional y los permisos, consulte Cuadernos.

Define las tablas con los decoradores de pipelines

Puedes definir una vista materializada independiente o una tabla de streaming con los mismos decoradores que usas en una canalización de Lakeflow. Cada función decorada define una tabla. Cuando ejecutas la celda, Azure Databricks crea la tabla y ejecuta un pipeline sin servidor para poblarla. La celda vuelve a aparecer cuando finaliza la actualización.

Warning

Los decoradores de canalizaciones requieren la versión 5 o superior del entorno serverless.

Defina una visión materializada

Úsala @dp.materialized_view en una función que devuelva un DataFrame por lotes. El siguiente ejemplo crea la vista daily_booking_revenue materializada a partir de la bookings tabla del conjunto de datos de ejemplo de 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 una tabla a partir de una lectura en streaming, use @dp.table en su lugar.

Definamos una tabla de streaming

Use @dp.table en una función que devuelve un DataFrame de streaming. El siguiente ejemplo crea la tabla de streaming bookings_raw a partir de una lectura en streaming de la misma tabla 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")

Si la función devuelve un DataFrame por lotes, @dp.table crea una vista materializada en su lugar. La única excepción es replace_where, que siempre da lugar a una tabla de streaming. El siguiente ejemplo mantiene actualizados los ingresos diarios de las visitas a partir del 1 de julio de 2025, sin necesidad de recalcular fechas 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 ejecución elimina las filas que coinciden con el predicado y recalcula solo ese rango. Consulte Procesamiento por lotes con flujos de tipo REPLACEWHERE.

Actualizar una tabla

Para refrescar una tabla que hayas definido con un decorador, ejecuta de nuevo el código que la define, por ejemplo volviendo a ejecutar la celda del cuaderno, ejecutando todo el cuaderno o ejecutándolo como un trabajo. Cada ejecución crea la tabla si no existe y la actualiza si ya existe.

Para reprocesar todos los datos disponibles en la fuente, pasa full_refresh=True a cualquiera de los 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")

No puedes usar una instrucción REFRESH sobre una tabla definida con un decorador ni programar actualizaciones con SCHEDULE o TRIGGER ON UPDATE. Para actualizar un calendario, define la tabla en SQL o programa el cuaderno como un trabajo. Consulte Trabajos de Lakeflow.

Configurar la tabla

Los decoradores aceptan los mismos parámetros comunes de conjunto de datos que aceptan dentro de una pipeline, incluyendo comment, table_properties, partition_cols, cluster_by, schema, y 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 la lista de parámetros, véase materialized_view y tabla.

private=True no está soportada, porque una tabla privada solo puede ser leída por otros conjuntos de datos en la misma canalización.

APIs no admitidas

Una tabla independiente es un único conjunto de datos con un único flujo, por lo que las APIs que describen las relaciones entre conjuntos de datos no están disponibles. Los siguientes generan error fuera de una canalización:

  • @dp.temporary_view y dp.create_streaming_table
  • @dp.append_flow y otros caudales adicionales
  • dp.create_auto_cdc_flow y dp.create_auto_cdc_from_snapshot_flow
  • @dp.replace_flow y el parámetro replace_using, que definen los flujos REPLACE USING. Consulte Reemplazo parcial de instantáneas con flujos REPLACE USING.
  • dp.create_sink
  • Expectativas, como @dp.expect y @dp.expect_or_fail

Para usar estos, crea en su lugar una tubería Lakeflow. Consulte Desarrollo de código de canalización con Python.

Enviar sentencias SQL con spark.sql()

En un cuaderno de Python, pase las mismas instrucciones que ejecutaría desde un almacenamiento de SQL de Databricks a spark.sql(). La sintaxis de la vista materializada independiente y de la tabla de streaming es idéntica; solo cambia la forma de enviar la instrucción. Al igual que con un almacén, cada instrucción CREATE o REFRESH ejecuta un pipeline sin servidor para procesar la operación.

La spark sesión está disponible de forma predeterminada en Azure Databricks cuadernos, por lo que no se requiere ninguna importación.

Creación de una vista materializada

En el ejemplo siguiente se crea la vista materializada mv1 a partir de la tabla 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 obtener detalles completos CREATE MATERIALIZED VIEW , como las actualizaciones programadas y desencadenadas, consulte Creación de una vista materializada.

Crear una tabla de streaming

En el ejemplo siguiente se crea la tabla de transmisión sales a partir de la tabla raw_data:

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

Para obtener información completa CREATE STREAMING TABLE , incluida la carga de archivos con el cargador automático y la programación, consulte Uso de tablas de streaming independientes.

Actualizar una vista materializada o una tabla de streaming

Use una REFRESH instrucción para actualizar una tabla independiente con los datos más recientes de su origen:

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

En el proceso general sin servidor, las actualizaciones son sincrónicas. No se admiten las actualizaciones asincrónicas (la ASYNC palabra clave). Consulte Proceso general sin servidor.

Parametrizar sentencias

Para pasar valores del código de Python a una instrucción en lugar de codificarlos de forma rígida, use marcadores de parámetros con nombre en SQL y proporcione sus valores a través del args argumento de spark.sql(). Use un marcador como :min_sales directamente para los valores literales. Encapsula el marcador en IDENTIFIER() solo cuando el parámetro sea un nombre de objeto, como una tabla, vista o esquema, porque los identificadores no se pueden sustituir como valores de cadena sin formato.

En el ejemplo siguiente se parametrizan el nombre de vista materializado y un 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 obtener más información, consulte Marcadores de parámetro y la IDENTIFIER cláusula.

Ejecutar otras instrucciones

Puede ejecutar cualquier instrucción de vista materializada independiente o de tabla de streaming desde un cuaderno de Python pasandola a spark.sql(), incluidas instrucciones para programar actualizaciones, modificar una tabla o quitar una tabla. Para comprender cómo usar vistas materializadas y tablas de streaming, incluida la sintaxis SQL, consulte Uso de vistas materializadas independientes y Uso de tablas de streaming independientes.

Limitaciones

Las vistas materializadas independientes y las tablas de streaming creadas en el entorno de proceso general sin servidor tienen limitaciones adicionales, como la falta de compatibilidad con los refrescos asíncronos y la ausencia de atribución de costes por tabla. Para obtener la lista completa, consulte Proceso general sin servidor.

Como estas canalizaciones se ejecutan en cómputo general sin servidor en lugar de en un almacén de datos SQL, no heredan etiquetas personalizadas de un almacén de datos circundante. La propagación de etiquetas de almacén a system.billing.usage se aplica solo a vistas materializadas y tablas de streaming cuyas sentencias se ejecutan desde un almacén SQL. Consulte Costos de atributos en SQL Warehouse con etiquetas personalizadas.

Las tablas definidas con los decoradores de canalizaciones tienen las siguientes limitaciones adicionales:

  • No puedes actualizarlos con una instrucción REFRESH ni programar actualizaciones con SCHEDULE o TRIGGER ON UPDATE. Ver Actualizar una tabla.
  • No se admiten expectativas, flujos adicionales, flujos de captura de datos por cambios (CDC), sumideros ni vistas temporales. Consulta APIs no soportadas.
  • No se admite private=True.

Recursos adicionales

  • Se actualizó por última vez el