Aller au contenu principal
Dernière mise à jour le

Développer le code des LakeFlow Pipelines avec SQL

LakeFlow Pipelines introduit plusieurs nouveaux mots-clés SQL et fonctions pour définir des vues matérialisées et des tables de streaming dans les pipelines. Le support SQL pour le développement de pipelines s'appuie sur les bases de Spark SQL et ajoute le support pour la fonctionnalité Structured Streaming.

Les utilisateurs familiarisés avec les DataFrames PySpark pourraient préférer développer du code de pipeline avec Python. Python prend en charge des tests et des opérations plus étendus qu'il est difficile de mettre en œuvre avec SQL, telles que les opérations de métaprogrammation. Consultez Développer le code de pipeline avec Python. Pour vous aider à choisir entre les deux interfaces, consultez Choisir entre SQL et Python.

Pour une référence complète de la syntaxe SQL de pipeline, consultez la référence du langage SQL de pipeline.

Principes de base du SQL pour le développement de pipelines​

Le code SQL qui crée des datasets de pipeline utilise la syntaxe CREATE OR REFRESH pour définir des vues matérialisées et des tables de streaming par rapport aux résultats de la query.

Le mot-clé STREAM indique si la source de données référencée dans une clause SELECT doit être lue avec une sémantique de streaming.

Les lectures et les écritures sont default le catalogue et le schéma spécifiés lors de la configuration du pipeline. Voir Définir le catalogue et le schéma cibles.

Le code source du pipeline diffère considérablement des scripts SQL : les LakeFlow Pipelines évaluent toutes les définitions de dataset dans tous les fichiers de code source configurés dans un pipeline et construisent un graphe de flux de données avant l'exécution de toute query. L’ordre des queries apparaissant dans les fichiers sources définit l’ordre d’évaluation du code, mais pas l’ordre d’exécution des queries.

Créer une vue matérialisée avec SQL​

L’exemple de code suivant présente la syntaxe de base pour la création d’une vue matérialisée avec SQL :

SQL
CREATE OR REFRESH MATERIALIZED VIEW basic_mv
AS SELECT * FROM samples.nyctaxi.trips;

Créer une table de streaming avec SQL​

L'exemple de code suivant démontre la syntaxe de base pour la création d'une table de streaming avec SQL. Lors de la lecture d'une source pour une table de streaming, le mot-clé STREAM indique d'utiliser la sémantique de streaming pour la source. N'utilisez pas le mot-clé STREAM lors de la création d'une vue matérialisée :

SQL
CREATE OR REFRESH STREAMING TABLE basic_st
AS SELECT * FROM STREAM samples.nyctaxi.trips;
remarque

Utilisez le mot-clé STREAM pour utiliser la sémantique de streaming afin de lire à partir de la source. Si la lecture rencontre une modification ou une suppression d'un enregistrement existant, une erreur est générée. Il est plus sûr de lire à partir de sources statiques ou en mode ajout uniquement. Pour ingérer les données qui ont des commits de modification, vous pouvez utiliser l'option skipChangeCommits pour gérer les erreurs.

Exemple :

SQL
CREATE OR REFRESH STREAMING TABLE basic_st
AS SELECT * FROM STREAM samples.nyctaxi.trips WITH (SKIPCHANGECOMMITS);

Charger des données depuis le stockage d'objets​

Les pipelines prennent en charge le chargement de données à partir de tous les formats pris en charge par Databricks. Consultez Options de format de données.

remarque

Ces exemples utilisent les données disponibles sous /databricks-datasets et montées automatiquement sur votre workspace. Databricks recommande d'utiliser des chemins de volume ou des URI cloud pour référencer les données stockées dans le stockage d'objets cloud. Consultez Que sont les volumes Unity Catalog ?.

Databricks recommande d'utiliser Auto Loader et les tables de streaming lors de la configuration des charges de travail d'ingestion incrémentielle pour les données stockées dans le stockage d'objets cloud. Consultez Qu'est-ce qu'Auto Loader ?.

SQL utilise la fonction read_files pour appeler la fonctionnalité Auto Loader. Vous devez également utiliser le mot-clé STREAM pour configurer une lecture en streaming avec read_files.

Ce qui suit décrit la syntaxe pour read_files en SQL :

CREATE OR REFRESH STREAMING TABLE table_name
AS SELECT *
  FROM STREAM read_files(
    "<file-path>",
    [<option-key> => <option_value>, ...]
  )

Les options pour Auto Loader sont des paires clé-valeur. Pour plus de détails sur les formats et options pris en charge, consultez Options.

L'exemple suivant crée une table de streaming à partir de fichiers JSON à l'aide d'Auto Loader :

SQL
CREATE OR REFRESH STREAMING TABLE ingestion_st
AS SELECT *
FROM STREAM read_files(
  "/databricks-datasets/retail-org/sales_orders",
  format => "json");

La fonction read_files prend également en charge la sémantique par batch pour créer des vues matérialisées. L’exemple suivant utilise la sémantique par batch pour lire un répertoire JSON et créer une vue matérialisée :

SQL
CREATE OR REFRESH MATERIALIZED VIEW batch_mv
AS SELECT *
FROM read_files(
  "/databricks-datasets/retail-org/sales_orders",
  format => "json");

Valider les données avec des attentes​

Vous pouvez utiliser des attentes pour définir et appliquer des contraintes de qualité des données. Consultez Gérer la qualité des données avec les attentes de pipeline.

Le code suivant définit une attente nommée valid_data qui supprime les enregistrements nuls lors de l'ingestion de données :

SQL
CREATE OR REFRESH STREAMING TABLE orders_valid(
  CONSTRAINT valid_date
  EXPECT (order_datetime IS NOT NULL AND length(order_datetime) > 0)
  ON VIOLATION DROP ROW
)
AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/sales_orders");

Query les vues matérialisées et les tables de streaming définies dans votre pipeline​

L'exemple suivant définit quatre datasets :

  • Une table de streaming nommée orders qui charge les données JSON.
  • Une vue matérialisée nommée customers qui charge les données CSV.
  • Une vue matérialisée nommée customer_orders qui joint les enregistrements des datasets orders et customers, convertit le timestamp de la commande en date, et sélectionne les champs customer_id, order_number, state et order_date.
  • Une vue matérialisée nommée daily_orders_by_state qui agrège le décompte quotidien des commandes pour chaque État.
remarque

Lorsque vous interrogez des vues ou des tables dans votre pipeline, vous pouvez spécifier le catalogue et le schéma directement, ou vous pouvez utiliser les valeurs default configurées dans votre pipeline. Dans cet exemple, les tables orders, customers et customer_orders sont écrites et lues à partir du catalogue et du schéma default configurés pour votre pipeline.

Le mode de publication hérité utilise le schéma LIVE pour query d'autres vues matérialisées et tables de streaming définies dans votre pipeline. Dans les nouveaux pipelines, la syntaxe de schéma LIVE est ignorée silencieusement. Consultez le schéma EN DIRECT (hérité).

SQL
CREATE OR REFRESH STREAMING TABLE orders(
  CONSTRAINT valid_date
  EXPECT (order_datetime IS NOT NULL AND length(order_datetime) > 0)
  ON VIOLATION DROP ROW
)
AS SELECT * FROM STREAM read_files("/databricks-datasets/retail-org/sales_orders");

CREATE OR REFRESH MATERIALIZED VIEW customers
AS SELECT * FROM read_files("/databricks-datasets/retail-org/customers");

CREATE OR REFRESH MATERIALIZED VIEW customer_orders
AS SELECT
  c.customer_id,
  o.order_number,
  c.state,
  date(timestamp(int(o.order_datetime))) order_date
FROM orders o
INNER JOIN customers c
ON o.customer_id = c.customer_id;

CREATE OR REFRESH MATERIALIZED VIEW daily_orders_by_state
AS SELECT state, order_date, count(*) order_count
FROM customer_orders
GROUP BY state, order_date;

Définir une table privée​

Vous pouvez utiliser la clause PRIVATE lors de la création d'une vue matérialisée ou d'une table en streaming. Lorsque vous créez une table privée, vous créez la table, mais pas les métadonnées de la table. La clause PRIVATE charge le pipeline de créer une table qui est disponible pour le pipeline, mais qui ne devrait pas être consultée en dehors du pipeline. Pour réduire le temps de traitement, une table privée persiste pendant toute la durée de vie du pipeline qui la crée, et non pas seulement pour une seule mise à jour.

Les tables privées peuvent avoir le même nom que les tables du catalogue. Si vous spécifiez un nom non qualifié pour une table dans un pipeline, s'il existe à la fois une table privée et une table de catalogue portant ce nom, la table privée est utilisée.

Les tables privées étaient auparavant appelées tables temporaires.

Supprimer définitivement les enregistrements d'une vue matérialisée ou d'une table de streaming​

Pour supprimer définitivement des enregistrements d'une table de streaming avec des vecteurs de suppression activés, par exemple pour la conformité GDPR, des opérations supplémentaires doivent être effectuées sur les tables Delta sous-jacentes de l'objet. Pour garantir la suppression des enregistrements d'une table de streaming, consultez Supprimer définitivement les enregistrements d'une table de streaming.

Les vues matérialisées reflètent toujours les données des tables sous-jacentes lorsqu'elles sont actualisées. Pour supprimer des données dans une vue matérialisée, vous devez supprimer les données de la source et refresh la vue matérialisée.

Paramétrez les valeurs utilisées lors de la déclaration de tables ou de vues avec SQL​

Utilisez SET pour spécifier une valeur de configuration dans une query qui déclare une table ou une vue, y compris les configurations Spark. Toute table ou vue que vous définissez dans un fichier source après l'instruction SET a accès à la valeur définie. Toutes les configurations Spark spécifiées à l'aide de l'instruction SET sont utilisées lors de l'exécution de la query Spark pour toute table ou vue suivant l'instruction SET. Pour lire une valeur de configuration dans une query, utilisez la syntaxe d'interpolation de chaînes ${}. L'exemple suivant définit une valeur de configuration Spark nommée startDate et utilise cette valeur dans une query :

SET startDate='2025-01-01';

CREATE OR REFRESH MATERIALIZED VIEW filtered
AS SELECT * FROM src
WHERE date > ${startDate}

Pour spécifier plusieurs valeurs de configuration, utilisez une instruction SET distincte pour chaque valeur.

Limitations​

La clause PIVOT n'est pas prise en charge. L'pivot opération dans Spark nécessite le chargement anticipé des données d'entrée pour calculer le schéma de sortie. Cette fonctionnalité n'est pas prise en charge dans les pipelines.

remarque

La syntaxe CREATE OR REFRESH LIVE TABLE pour créer une vue matérialisée est dépréciée. Utilisez plutôt CREATE OR REFRESH MATERIALIZED VIEW.

Sur cette page