Utiliser Python avec des pipelines autonomes

Important

La création et l’actualisation de vues matérialisées autonomes et de tables de streaming depuis un notebook sur un calcul serverless à usage général est en version bêta et disponible que dans certaines régions. Voir Notebooks.

Vous pouvez créer et actualiser des vues matérialisées autonomes et diffuser des tables à partir d’un bloc-notes à l’aide de Python. Cela vous permet de gérer des pipelines autonomes en même temps que vos autres workflows de notebooks basés sur Python.

Il existe deux méthodes pour y parvenir :

  • Définissez la table à l’aide des décorateurs pyspark.pipelines, @dp.materialized_view et @dp.table. Utilisez cela lorsque la logique est plus facile à exprimer sous forme de code DataFrame. Voir Définir des tables à l’aide des décorateurs de pipeline.
  • Envoyez à spark.sql() les mêmes instructions SQL que celles qu’exécute un entrepôt SQL Databricks. Cela vous donne l'intégralité de la prise en charge SQL des vues matérialisées autonomes et des tables en streaming, y compris les instructions REFRESH et les calendriers d'actualisation. Voir Soumettre des instructions SQL avec spark.sql().

Le code source Python des pipelines autonomes nécessite un notebook associé à un calcul serverless à usage général. Vous ne pouvez pas utiliser Python pour créer ou actualiser des pipelines autonomes à partir d'un entrepôt SQL Databricks, car un entrepôt exécute des instructions SQL, et non Python notebooks. Pour utiliser un entrepôt SQL à la place, consultez Utiliser des vues matérialisées autonomes et utiliser des tables de diffusion en continu autonomes.

Requirements

Pour créer et mettre à jour des pipelines autonomes avec Python, vous devez disposer d’un notebook associé à un calcul serverless à usage général sur Databricks Runtime 18.1 ou version supérieure. Pour obtenir la liste complète des exigences, notamment la disponibilité régionale et les autorisations, consultez Notebooks.

Définissez des tables à l’aide des décorateurs de pipeline

Vous pouvez définir une vue matérialisée autonome ou une table de flux avec les mêmes décorateurs que ceux que vous utilisez dans un pipeline Lakeflow. Chaque fonction décorée définit une table. Lorsque vous exécutez la cellule, Azure Databricks crée la table et exécute un pipeline sans serveur pour l’alimenter. La cellule réapparaît une fois la mise à jour terminée.

Warning

Les décorateurs de pipeline nécessitent la version de l’environnement serverless 5 ou une version ultérieure.

Définir une vue matérialisée

Utilisez @dp.materialized_view sur une fonction qui retourne un DataFrame de lots. L’exemple suivant crée la vue daily_booking_revenue matérialisée à partir du bookings tableau dans le jeu de données d’exemple 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"))
  )

Pour définir une table à partir d’une lecture en streaming, utilisez @dp.table à la place.

Définir une table de diffusion en continu

À utiliser @dp.table sur une fonction qui renvoie un DataFrame en streaming. L’exemple suivant crée la table de streaming bookings_raw à partir d’une lecture de streaming de la même table 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 fonction renvoie un DataFrame par lots, @dp.table crée à la place une vue matérialisée. La seule exception est replace_where, qui entraîne toujours la création d’une table de streaming. L’exemple suivant maintient à jour les revenus quotidiens des check-ins à partir du 1er juillet 2025, sans recalculer les dates antérieures :

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

Chaque exécution supprime les lignes correspondant au prédicat et ne recalcule que cette plage. Voir le traitement par lots avec des flux REPLACE WHERE.

Actualiser un tableau

Pour rafraîchir une table que vous avez définie avec un décorateur, exécutez à nouveau le code qui la définit, par exemple en relançant la cellule du carnet, en exécutant l’intégralité du carnet, ou en exécutant le carnet en tant que travail. Chaque exécution crée la table si elle n’existe pas et la rafraîchit si elle existe.

Pour retraiter toutes les données disponibles dans la source, passez full_refresh=True à l’un ou l’autre des décorateurs :

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

Vous ne pouvez pas utiliser une instruction REFRESH sur une table définie à l’aide d’un décorateur, ni planifier des actualisations avec SCHEDULE ou TRIGGER ON UPDATE. Pour effectuer une actualisation planifiée, définissez la table en SQL ou planifiez le notebook en tant que tâche. Consultez les offres d'emploi Lakeflow.

Configurez la table

Les décorateurs acceptent les mêmes paramètres communs de jeu de données qu’ils acceptent dans un pipeline, notamment comment, table_properties, partition_cols, cluster_by, schema, et 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"))
  )

Pour la liste des paramètres, voir materialized_view et tableau.

private=True n’est pas prise en charge, car une table privée ne peut être lue que par d’autres jeux de données dans le même pipeline.

APIs non supportées

Une table autonome est un ensemble de données unique avec un seul flux, donc les API qui décrivent les relations entre ensembles de données ne sont pas disponibles. Les éléments suivants provoquent une erreur en dehors d’un pipeline :

  • @dp.temporary_view et dp.create_streaming_table
  • @dp.append_flow et d'autres flux supplémentaires
  • dp.create_auto_cdc_flow et dp.create_auto_cdc_from_snapshot_flow
  • @dp.replace_flow et le paramètre replace_using, qui définissent les flux REPLACE USING. Voir Remplacement partiel d’instantané à l’aide des flux REPLACE USING.
  • dp.create_sink
  • Attentes, telles que @dp.expect et @dp.expect_or_fail

Pour les utiliser, créez plutôt un pipeline Lakeflow. Consultez Développer du code de pipeline avec Python.

Soumettre des instructions SQL avec spark.sql()

Dans un notebook Python, passez les mêmes instructions que celles que vous exécutez à partir d’un entrepôt Databricks SQL vers spark.sql(). La syntaxe de la vue matérialisée autonome et de la table de diffusion en continu est identique ; seule la manière dont vous soumettez l’instruction diffère. Comme avec un entrepôt, chaque CREATE ou REFRESH instruction exécute un pipeline serverless pour traiter l’opération.

La spark session est disponible par défaut dans Azure Databricks notebooks, donc aucune importation n’est requise.

Créer une vue matérialisée

L’exemple suivant crée la vue matérialisée mv1 à partir de la table de 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
""")

Pour plus CREATE MATERIALIZED VIEW d’informations, telles que les actualisations planifiées et déclenchées, consultez Créer une vue matérialisée.

Créer une table de diffusion en continu

L’exemple suivant crée la table de diffusion en continu sales à partir de la table raw_data :

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

Pour plus CREATE STREAMING TABLE d’informations, notamment le chargement de fichiers avec le chargeur automatique et la planification, consultez Utiliser des tables de diffusion en continu autonomes.

Rafraîchir une vue matérialisée ou une table de streaming

Utilisez une REFRESH instruction pour mettre à jour une table autonome avec les données les plus récentes de sa source :

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

Dans le calcul générique serverless, les actualisations sont synchrones. Les actualisations asynchrones (le ASYNC mot clé) ne sont pas prises en charge. Consultez le calcul généraliste sans serveur.

Paramétriser des instructions

Pour passer des valeurs de votre code Python dans une instruction au lieu de les coder en dur, utilisez des marqueurs de paramètres nommés dans sql et fournissez leurs valeurs par le biais de l’argument args de spark.sql(). Utilisez un marqueur tel que :min_sales directement pour les valeurs littérales. Encapsulez le marqueur IDENTIFIER() uniquement lorsque le paramètre est un nom d’objet, tel qu’une table, une vue ou un schéma, car les identificateurs ne peuvent pas être substitués en tant que valeurs de chaîne simple.

L’exemple suivant paramétre à la fois le nom de la vue matérialisée et une valeur de filtre :

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

Pour plus d’informations, consultez les marqueurs de paramètre et IDENTIFIER la clause.

Exécuter d’autres instructions

Vous pouvez exécuter toute instruction autonome sur une vue matérialisée ou une table de streaming à partir d’un notebook Python en la passant à spark.sql(), y compris les instructions permettant de planifier des rafraîchissements, de modifier une table ou de supprimer une table. Pour comprendre comment utiliser des vues matérialisées et des tables de diffusion en continu, notamment la syntaxe SQL, consultez Utiliser des vues matérialisées autonomes et utiliser des tables de diffusion en continu autonomes.

Limitations

Les vues matérialisées autonomes et les tables de streaming créées sur du calcul serverless à usage général présentent des limitations supplémentaires, notamment l’absence de prise en charge des actualisations asynchrones et de l’attribution des coûts par table. Pour consulter la liste complète, voir calcul général sans serveur.

Parce que ces pipelines s’exécutent sur du calcul général sans serveur plutôt que sur un entrepôt SQL, ils n’héritent pas des balises personnalisées d’un entrepôt englobant. La propagation des tags d’entrepôt à system.billing.usage s’applique uniquement aux vues matérialisées et aux tables de streaming dont les instructions s’exécutent dans un entrepôt SQL. Consultez Attribuer les coûts au SQL Warehouse à l’aide de balises personnalisées.

Les tables définies avec les décorateurs de pipeline sont soumises aux limitations supplémentaires suivantes :

  • Vous ne pouvez pas les actualiser avec une instruction REFRESH, ni planifier des actualisations avec SCHEDULE ou TRIGGER ON UPDATE. Voir Rafraîchir un tableau.
  • Les contraintes, les flux supplémentaires, les flux de capture des données modifiées (CDC), les sinks et les vues temporaires ne sont pas pris en charge. Voir les API non prises en charge.
  • La fonction private=True n'est pas prise en charge.

Ressources additionnelles

  • Dernière mise à jour :