Aller au contenu principal
Dernière mise à jour le

Utiliser Python avec des pipelines autonomes

info

Bêta

La création et l’actualisation de vues matérialisées et de tables de streaming autonomes à partir d’un notebook sur un compute général serverless sont en bêta et disponibles dans certaines régions. Voir les Notebooks.

Vous pouvez créer et refresh des vues matérialisées et des tables de streaming autonomes à partir d'un Notebook à l'aide de Python. Cela vous permet de gérer des pipelines autonomes avec vos autres workflows de notebooks basés sur Python.

Pour ce faire, deux méthodes sont possibles :

  • Définissez la table avec les décorateurs pyspark.pipelines, @dp.materialized_view et @dp.table. À utiliser lorsque la logique est plus facile à exprimer sous forme de code DataFrame. Voir Définir des tables avec les décorateurs de pipeline.
  • Soumettez les mêmes instructions SQL qu'un Databricks SQL Warehouse exécute en les transmettant à spark.sql(). Cela vous donne la surface SQL complète pour les vues matérialisées et les tables de streaming autonomes, y compris les instructions REFRESH et les programmes de refresh. Consultez Soumettre des instructions SQL avec spark.sql().

Le code source Python pour les pipelines autonomes nécessite un notebook connecté au compute général serverless . Vous ne pouvez pas utiliser Python pour créer ou refresh des pipelines autonomes à partir d’un warehouse Databricks SQL, car un warehouse exécute des instructions SQL, pas des Notebook Python. Pour utiliser un SQL warehouse à la place, voir Utiliser des vues matérialisées autonomes et Utiliser des tables streaming autonomes.

Prérequis​

Pour créer et refresh des pipelines autonomes avec Python, vous avez besoin d’un Notebook connecté au compute general Serverless sur Databricks Runtime 18.1 ou version supérieure. Pour obtenir la liste complète des prérequis, notamment la disponibilité régionale et les autorisations, consultez la section Notebooks.

Définir des tables avec les décorateurs de pipelines​

Vous pouvez définir une vue matérialisée ou une table de streaming autonome avec les mêmes décorateurs que ceux utilisés dans un LakeFlow Pipelines. Chaque fonction décorée définit une table. Lorsque vous exécutez la cellule, Databricks crée la table et exécute un pipeline serverless pour la remplir. La cellule renvoie une valeur lorsque la mise à jour est terminée.

attention

Les décorateurs de pipelines nécessitent la version 5 ou supérieure de l’environnement serverless.

Définir une vue matérialisée​

Utilisez @dp.materialized_view sur une fonction qui renvoie un DataFrame batch. L’exemple suivant montre comment créer la vue matérialisée daily_booking_revenue à partir de la table bookings dans le dataset d’exemple 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"))
  )

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

Définir une table de streaming​

Utilisez @dp.table sur une fonction qui renvoie un DataFrame de streaming. L'exemple suivant crée la table de streaming bookings_raw à partir d'une lecture en streaming de la même table 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")

Si la fonction renvoie un DataFrame batch, @dp.table crée plutôt une vue matérialisée. La seule exception est replace_where, qui donne toujours lieu à une table de streaming. L'exemple suivant maintient à jour le revenu quotidien des enregistrements effectués à partir du 1er juillet 2025, sans recalculer les dates précédentes :

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

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

refresh une table​

Pour refresh une table que vous avez définie avec un décorateur, réexécutez le code qui la définit, par exemple en réexécutant la cellule du Notebook, en exécutant l’intégralité du Notebook ou en exécutant le Notebook en tant que job. Chaque exécution crée la table si elle n’existe pas et la refresh si elle existe.

Pour traiter à nouveau toutes les données disponibles dans la source, passez full_refresh=True sur l’un ou l’autre décorateur :

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

Vous ne pouvez pas utiliser une instruction REFRESH sur une table définie avec un décorateur, ou planifier des refresh avec SCHEDULE ou TRIGGER ON UPDATE. Pour refresh selon un calendrier, définissez la table en SQL ou planifiez le Notebook en tant que Job. Voir Lakeflow Jobs.

Configurer la table​

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

Pour la liste des paramètres, consultez materialized_view et table.

private=True n'est pas pris 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 prises en charge​

Une table autonome est un dataset unique avec un flux unique, de sorte que les APIs qui décrivent les relations entre les datasets ne sont pas disponibles. Ce qui suit génère 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 parameter replace_using, qui définit les flux REPLACE USING. Voir Remplacement partiel de snapshot avec 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 Pipelines. Voir Développer le code de pipeline avec Python.

Envoyer des instructions SQL avec spark.sql()​

Dans un Notebook Python, transmettez les mêmes instructions que celles que vous exécuteriez depuis un warehouse Databricks SQL vers spark.sql(). La syntaxe des vues matérialisées et des tables de streaming autonomes est identique ; seule la méthode de soumission de l’instruction diffère. Comme pour un warehouse, chaque instruction CREATE ou REFRESH exécute un pipeline serverless pour traiter l’opération.

La session spark est disponible par default dans les notebooks Databricks ; aucune importation n’est donc requise.

Créer une vue matérialisée​

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

Pour obtenir des détails complets sur CREATE MATERIALIZED VIEW, comme les refresh planifiés et Trigger, voir Créer une vue matérialisée.

Créer une table de streaming​

L'exemple suivant crée la table de streaming sales à partir de la table raw_data :

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

Pour tous les détails sur CREATE STREAMING TABLE, y compris le chargement des fichiers avec Auto Loader et la planification, voir Utiliser des tables streaming autonomes.

refresh une vue matérialisée ou une table de streaming​

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

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

Sur un compute général Serverless, les refresh sont synchrones. Les refresh asynchrones (mot-clé ASYNC) ne sont pas prises en charge. Voir Compute général serverless.

Paramétrer des instructions​

Pour transmettre des valeurs de votre code Python à une instruction au lieu de les coder en dur, utilisez des marqueurs de paramètres nommés dans le 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. Placez le marqueur entre IDENTIFIER() uniquement lorsque le parameter est un nom d'objet, tel qu'une table, une vue ou un schéma, car les identifiants ne peuvent pas être substitués par des valeurs de chaîne simples.

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

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

Pour plus d’information, voir Marqueurs de parameter et clause IDENTIFIER.

Exécuter d’autres instructions​

Vous pouvez exécuter n’importe quelle instruction de vue matérialisée ou de table de streaming autonome à partir d’un notebook Python en la transmettant à spark.sql(), notamment les instructions pour planifier des refresh, modifier une table ou supprimer une table. Pour comprendre comment utiliser les vues matérialisées et les tables de streaming, y compris la syntaxe SQL, voir Utiliser des vues matérialisées autonomes et Utiliser des tables de streaming autonomes.

Limitations​

Les vues matérialisées autonomes et les tables streaming créées sur un compute général Serverless ont des limitations supplémentaires, comme l’absence de prise en charge des refresh asynchrones et l’absence d’attribution des coûts par table. Pour la liste complète, voir Compute général serverless.

Comme ces pipelines s’exécutent sur le compute général serverless plutôt que sur un SQL warehouse, ils n’héritent pas des tags personnalisés d’un warehouse englobant. La propagation des tags de warehouse à system.billing.usage s’applique uniquement aux vues matérialisées et aux tables de streaming dont les instructions s’exécutent à partir d’un SQL warehouse. Consultez Attribuer des coûts au SQL warehouse avec des tags personnalisés.

Les tables définies avec les décorateurs de pipelines présentent les limites supplémentaires suivantes :

  • Vous ne pouvez pas les refresh avec une instruction REFRESH ou planifier des refresh avec SCHEDULE ou TRIGGER ON UPDATE. Voir refresh une table.
  • Les attentes, les flux supplémentaires, les flux de Change Data Capture (CDC), les puits et les vues temporaires ne sont pas pris en charge. Voir APIs non prises en charge.
  • private=True n’est pas pris en charge.

Ressources supplémentaires​

Sur cette page