Data Engineer Associate

1. L'examen en bref

Le nouveau référentiel ne publie pas de pondération par domaine. Il couvre :

1. Databricks Data Intelligence Platform ;
2. ingestion et chargement ;
3. transformation et modélisation ;
4. Lakeflow Jobs ;
5. CI/CD ;
6. dépannage, supervision et optimisation ;
7. gouvernance et sécurité.

L'examen teste surtout la capacité à choisir le bon outil pour un scénario, à reconnaître du code correct et à diagnostiquer un problème. Il faut donc connaître les différences entre les solutions, pas seulement mémoriser leur syntaxe.

2. Modèle mental de la plateforme

2.1 Lakehouse et Data Intelligence Platform

Un lakehouse combine :

• la souplesse et le coût du stockage objet d'un data lake ;
• les transactions, le schéma, les performances et les usages BI d'un data warehouse ;
• une gouvernance commune pour les données et les actifs d'IA.

Les briques à retenir sont :

Le stockage et le calcul sont séparés. Les tables métier résident généralement dans un stockage objet durable ; le compute démarre, lit les données, calcule, puis peut s'arrêter sans supprimer les données.

Pour les questions d'architecture :

• Databricks SQL est le service de data warehousing de la Data Intelligence Platform ; il s'appuie sur des SQL warehouses pour exécuter les requêtes SQL et servir les usages BI ;
• l'application web du workspace et les services backend managés se trouvent dans le control plane ;
• le compute plane exécute le traitement des données, sur compute classic ou serverless selon la configuration.

La phrase de QCM « les données client sont entièrement hébergées dans le compte cloud du client » décrit au mieux une architecture classic simplifiée : le classic compute et le workspace storage sont dans le compte cloud du client. Elle n'est pas universelle : le serverless compute s'exécute dans un compute plane géré par Databricks, et certains objets workspace ou résultats sont gérés dans le control plane/default storage. Si l'énoncé précise l'architecture classic ou oppose le stockage métier au control plane, « compte cloud du client » reste la réponse attendue ; si l'énoncé inclut le serverless, il faut nuancer. Toujours distinguer où les tables sont stockées, où le code est conservé et où il s'exécute.

Où est stocké un notebook ?

Le source et les métadonnées d'un notebook Databricks sont des objets du workspace stockés dans le control plane, chiffrés au repos comme données de services managés. Le notebook envoie son code au compute, mais il n'est pas stocké dans les workers du compute plane avec les tables métier.

Lorsqu'un notebook est déclenché par un Job, son code s'exécute dans le compute plane, sur le compute associé à la tâche : cluster Spark classic, jobs compute ou compute serverless. Mémo : notebook stocké dans le control plane ; commandes exécutées dans le compute plane.

Ne pas généraliser cette réponse aux résultats : les petits résultats interactifs peuvent aussi être stockés dans le control plane, tandis que les résultats volumineux et ceux des notebooks exécutés comme Jobs sont généralement placés dans le stockage du workspace. L'emplacement physique de ce stockage dépend notamment du type de workspace — compte cloud client en classic, default storage géré en serverless — et de sa configuration. Pour la question « où est stocké le notebook ? », la réponse attendue est bien control plane, comme objet workspace sécurisé et chiffré.

2.2 Choisir le compute

Points de décision :

• Serverless : préférence lorsque la compatibilité du workload le permet ; Databricks gère l'infrastructure.
• Classic : contrôle du runtime, des types de machines, des bibliothèques et du réseau ; l'infrastructure se trouve dans le compte cloud du client.
• All-purpose : interactif et partagé.
• Job compute : automatisé et éphémère.
• SQL warehouse : SQL/BI, pas un cluster Spark généraliste.
• Photon : moteur vectorisé natif qui accélère de nombreuses requêtes SQL et DataFrame sans réécriture du code.
• Le coût total dépend des DBU, de la durée et, pour le classic compute, des ressources du fournisseur cloud. Éteindre ou auto-terminer les ressources inactives.

Sur un compute classic, le même type d'instance avec Photon consomme des DBU à un taux différent du runtime non-Photon, souvent supérieur selon le cloud et le SKU. Ce n'est toutefois pas un « surcoût Photon » universel : Photon est déjà activé/inclus sur les SQL warehouses et les computes serverless. Comparer le coût total du run, pas seulement les DBU par heure : une exécution plus chère à l'heure peut coûter moins si Photon réduit suffisamment sa durée. Vérifier les tarifs du contrat et les données de facturation plutôt que mémoriser un multiplicateur fixe.

Access modes : Standard ou Dedicated

L'access mode détermine l'isolation des utilisateurs et les fonctionnalités disponibles sur un compute classic gouverné par Unity Catalog :

Scénario de QCM : code exclusivement en R + Unity Catalog + packages système installés par un init script cluster-scoped → Dedicated. R suffit déjà à exclure Standard. Le script peut être stocké dans un volume Unity Catalog et attaché dans Compute > Advanced > Init Scripts ; il s'exécute au démarrage sur chaque nœud du cluster. Avec Standard, les init scripts et bibliothèques concernés doivent en plus respecter l'allowlist, ce qui ne résout de toute façon pas l'absence de support de R.

Les init scripts stockés dans DBFS root sont en fin de vie et ne fonctionnent plus. Préférer un volume Unity Catalog, un workspace file ou le stockage cloud selon le mode d'accès ; un init script issu d'un workspace file n'est notamment pas pris en charge en mode Standard. Utiliser d'abord les fonctions natives de la plateforme ou les bibliothèques déclarées, et réserver les init scripts aux dépendances réellement système.

Dedicated est le nouveau nom de l'ancien mode Single user et peut être affecté à un utilisateur ou, avec les prérequis de runtime, à un groupe. Ne pas le choisir systématiquement : utiliser Standard pour le partage compatible et serverless lorsque le workload le permet.

Instances Spot/Preemptible

Les workers Spot coûtent moins cher, mais le fournisseur cloud peut les reprendre avec peu de préavis. Le meilleur cas est donc un batch non critique, relançable et idempotent, avec retries et checkpointing si nécessaire. Éviter de leur confier seuls un flux continu, une tâche interactive ou un workload à SLA strict qui ne tolère pas la perte d'un worker.

Sur AWS, lorsque l'option Spot standard du compute est activée, le driver reste on-demand et les workers suivants peuvent être Spot. La tolérance aux pannes réduit le risque mais ne rend pas le compute Spot aussi fiable que l'on-demand.

Instance pools pour le compute classic

Un instance pool, désormais souvent appelé simplement pool, conserve un ensemble d'instances cloud inactives et prêtes à être affectées à des clusters. Un cluster classic rattaché au pool peut réutiliser ces machines au lieu d'attendre leur provisionnement : cela réduit son startup time et le délai d'autoscaling.

Pool d'instances prêtes
        ├── cluster Job A démarre et prend des instances
        ├── cluster Job B scale out et en prend d'autres
        └── à l'arrêt, les instances retournent au pool

Réglages à comprendre :

• Minimum idle instances : nombre d'instances maintenues prêtes ; plus il est élevé, plus le démarrage est rapide mais plus le coût cloud au repos augmente ;
• Maximum capacity : nombre maximal d'instances, actives et inactives, que le pool peut contenir ;
• Idle instance auto termination : durée après laquelle une instance inactive au-delà du minimum peut être libérée ;
• Preloaded Databricks Runtime : runtime préchargé sur les instances inactives pour réduire davantage le démarrage.

Si aucune instance idle compatible n'est disponible, le pool doit en demander une nouvelle au fournisseur cloud : le gain de démarrage diminue. Les instances inactives ne consomment pas de DBU, mais restent facturées par le fournisseur cloud. Les pools concernent le compute classic ; lorsqu'un workload est compatible, Databricks recommande généralement le serverless, dont l'infrastructure est gérée automatiquement.

Auto Stop d'un SQL warehouse

Auto Stop arrête le warehouse après une durée configurée sans requête active afin d'éviter de facturer un compute inactif. Une nouvelle requête, un Job, une connexion JDBC/ODBC ou l'ouverture d'un dashboard peut le redémarrer automatiquement, avec un éventuel délai de démarrage.

• Serverless : 10 minutes par défaut dans l'UI, minimum 5 minutes dans l'UI.
• Pro/classic : 45 minutes par défaut, minimum 10 minutes.
• Ne pas confondre Auto Stop, qui arrête un warehouse inactif, avec l'autoscaling, qui adapte le nombre de clusters à la concurrence pendant son fonctionnement.

Un délai court réduit le coût des usages sporadiques ; un délai plus long peut réduire les redémarrages pour un trafic intermittent. Le réglage se trouve dans SQL Warehouses > Edit > Auto Stop.

2.3 Unity Catalog : namespace et objets de données

Le nom complet d'une table suit la hiérarchie à trois niveaux :

catalog.schema.object

Exemple :

SELECT * FROM production.sales.orders;

• Le metastore est le conteneur de gouvernance de plus haut niveau.
• Un catalog sert souvent de frontière métier, d'environnement ou d'isolation.
• Un schema organise les objets d'un catalogue.
• Une table, une vue, un volume, une fonction ou un modèle est un objet sécurisable.
• Un volume gouverne des fichiers non tabulaires. Son chemin est :

/Volumes/<catalog>/<schema>/<volume>/<chemin>

Utiliser un nom de table pour les données tabulaires et un volume pour les fichiers. Éviter l'accès direct par chemin aux tables managées, car il contournerait la gouvernance Unity Catalog.

2.3.1 Storage credentials et external locations

Une storage credential est un objet sécurisable du metastore qui représente le mécanisme d'authentification au stockage cloud, par exemple un rôle IAM AWS, sans exposer ses secrets dans les notebooks. Elle ne désigne aucun chemin.

Une external location associe une URL/prefixe de stockage à une storage credential. Elle borne donc où cette identité peut être utilisée et permet à Unity Catalog d'appliquer des privilèges sur ce chemin. Elle ne copie pas les fichiers et n'est pas elle-même une table.

Relation à mémoriser :

Identité cloud (IAM role / managed identity / service account)
        -> Storage Credential (comment s'authentifier)
        -> External Location (quel chemin gouverner)
        -> External Table ou External Volume (quel objet de données exposer)

Une table gouverne des lignes interrogeables ; un volume gouverne des fichiers accessibles par chemin. Choisir un external volume lorsqu'un système externe dépose déjà des réponses REST, JSON, images ou autres fichiers non tabulaires dans un bucket géré par le client. Unity Catalog gouverne alors les accès effectués via /Volumes, mais un outil qui accède directement à l'URI cloud reste soumis aux contrôles IAM du cloud.

La storage credential est créée par un administrateur via Catalog Explorer, CLI, SDK ou API selon le cloud. L'external location et les objets de données s'expriment naturellement en DDL SQL :

CREATE EXTERNAL LOCATION IF NOT EXISTS sales_landing
URL 's3://company-data/sales'
WITH (STORAGE CREDENTIAL company_sales_credential)
COMMENT 'Zone de données Sales gouvernée par Unity Catalog';

Pour une API REST, un external volume peut constituer une bonne zone de fichiers bruts ; son exemple de création est placé avec les autres objets externes en 2.3.2.2. Si le JSON doit ensuite être filtré, joint ou agrégé, l'ingérer dans une table Delta Bronze/Silver. Un volume ne remplace donc pas une table tabulaire.

Il n'existe pas d'alternative DataFrame PySpark pour créer ces objets de gouvernance ; utiliser le DDL SQL, éventuellement envoyé par spark.sql(...). PySpark devient pertinent pour lire ou écrire un DataFrame dans une table déjà autorisée, comme dans la section suivante. Le droit cloud de la credential ne suffit pas : le principal Databricks doit aussi recevoir les privilèges Unity Catalog appropriés sur l'external location et les objets.

Sur Azure, le rôle Storage Blob Data Contributor est couramment attribué à l'access connector ou à l'identité managée pour permettre la lecture et l'écriture dans ADLS. Ce rôle n'est toutefois pas une option d'Auto Loader : c'est un droit du stockage. Une ingestion strictement en lecture peut recevoir un rôle plus restreint, tandis que la création de notifications/file events requiert aussi les permissions Azure Event Grid et de file events prévues par la configuration. Toujours distinguer droits cloud, storage credential et grants Unity Catalog.

2.3.2 Créer les catalogues, schémas et tables

Le type est déterminé au niveau de la table, pas du schéma : un même schéma standard Unity Catalog peut contenir des tables managées et externes. Une table étrangère appartient, elle, à un foreign catalog qui reflète une source distante.

2.3.2.1 Table managée

Une table managée nécessite un managed storage disponible au niveau du schéma, du catalogue ou du metastore. Si l'on déclare explicitement MANAGED LOCATION, son URL doit être couverte par une external location et le créateur doit y posséder CREATE MANAGED STORAGE.

-- Hypothèse : « company_managed_storage » couvre s3://company-uc/.
GRANT CREATE MANAGED STORAGE
ON EXTERNAL LOCATION company_managed_storage
TO `data_platform_admins`;

-- Définit le stockage par défaut des objets managés du catalogue.
CREATE CATALOG IF NOT EXISTS production
MANAGED LOCATION 's3://company-uc/production';

-- Hérite de la managed location du catalogue (sinon : MANAGED LOCATION 's3://company-uc/production/sales').
CREATE SCHEMA IF NOT EXISTS production.sales;

-- Sans LOCATION : Unity Catalog choisit le chemin, donc table managée.
CREATE TABLE production.sales.orders (
  order_id BIGINT,
  order_date DATE,
  amount DECIMAL(12, 2)
) USING DELTA;

CREATE CATALOG crée automatiquement default et information_schema, mais aucun schéma métier ni aucune table métier : production.sales et production.sales.orders doivent donc être créés explicitement. Si le metastore possède déjà un managed storage, le catalogue peut lui-même l'hériter et être créé sans MANAGED LOCATION.

Pour isoler les futurs objets de production.sales, on pourrait déclarer ce schéma avec MANAGED LOCATION 's3://company-uc/production/sales' à la place de la création par héritage présentée ci-dessus.

Ordre de résolution : schéma, sinon catalogue, sinon metastore. Unity Catalog génère des sous-répertoires internes uniques. Modifier ultérieurement une managed location ne déplace pas les objets existants ; le nouvel emplacement ne concerne que les nouveaux objets.

Version PySpark lorsque le DataFrame existe déjà :

orders_df.write.mode("overwrite").saveAsTable("production.sales.orders")

2.3.2.2 Table externe

Une table externe utilise un LOCATION explicite couvert par une external location, ici sales_landing créée en 2.3.1. Le catalogue et le schéma restent des objets standard ; leur éventuelle managed location n'est pas utilisée par cette table. La création suppose aussi USE CATALOG, USE SCHEMA et CREATE TABLE sur les parents.

-- Les droits sur le catalogue et le schéma sont supposés déjà accordés.
-- READ FILES est utile si l'équipe doit aussi lire directement les fichiers.
GRANT CREATE EXTERNAL TABLE, READ FILES
ON EXTERNAL LOCATION sales_landing TO `data_engineers`;

CREATE TABLE production.sales.orders_external (
  order_id BIGINT,
  order_date DATE,
  amount DECIMAL(12, 2)
) USING DELTA
LOCATION 's3://company-data/sales/orders';

Version PySpark :

(orders_df.write
          .format("delta")
          .mode("overwrite")
          .option("path", "s3://company-data/sales/orders")
          .saveAsTable("production.sales.orders_external"))

Le chemin externe ne doit pas chevaucher le managed storage du catalogue ou du schéma. DROP TABLE supprimera l'enregistrement Unity Catalog, mais conservera ces fichiers.

Un external volume gouverne des fichiers non tabulaires et non des lignes. Outre USE CATALOG et USE SCHEMA, il requiert CREATE VOLUME sur le schéma et CREATE EXTERNAL VOLUME sur l'external location :

CREATE SCHEMA IF NOT EXISTS production.raw;

GRANT CREATE VOLUME
ON SCHEMA production.raw TO `data_engineers`;

GRANT CREATE EXTERNAL VOLUME
ON EXTERNAL LOCATION sales_landing TO `data_engineers`;

CREATE EXTERNAL VOLUME production.raw.api_payloads
LOCATION 's3://company-data/sales/api-payloads';

Les consommateurs utilisent ensuite /Volumes/production/raw/api_payloads/... et reçoivent READ VOLUME ou WRITE VOLUME sur le volume selon leur besoin. READ FILES sur l'external location et READ VOLUME sur le volume sont des privilèges distincts.

2.3.2.3 Table étrangère

Une table étrangère n'est généralement pas créée comme une table locale. Pour Lakehouse Federation, une CONNECTION décrit la source et ses credentials, puis un FOREIGN CATALOG expose en lecture les schémas et tables qui existent dans le système distant :

CREATE CONNECTION operational_postgres
TYPE POSTGRESQL
OPTIONS (
  host 'postgres.example.com',
  port '5432',
  user secret ('federation', 'postgres_user'),
  password secret ('federation', 'postgres_password')
);

CREATE FOREIGN CATALOG operational
USING CONNECTION operational_postgres
OPTIONS (database 'orders');

SELECT * FROM operational.public.orders;

La CONNECTION ne copie aucune donnée et ne suffit pas à exposer les tables. Le foreign catalog reflète les objets de la source ; les foreign tables issues de Lakehouse Federation sont en lecture seule. Il n'existe pas d'équivalent DataFrame PySpark à ces DDL de gouvernance, mais un programme Python peut les exécuter avec spark.sql(...).

Synthèse :

Pour déclarer précisément les objets de gouvernance, SQL est généralement plus explicite. PySpark est naturel lorsque le DataFrame à écrire existe déjà dans le traitement.

Renommer une table managée ou externe est une opération de métadonnées : les données ne sont pas recopiées.

ALTER TABLE production.sales.orders_draft
RENAME TO production.sales.orders;

Dans Unity Catalog, la cible du renommage doit rester dans le même catalogue. Cette commande ne sert donc pas à déplacer une table entre catalogues ; mettre à jour les consommateurs qui utilisent l'ancien nom.

2.3.3 Convertir avec SET MANAGED, UNSET MANAGED, MOVE et COPY

Databricks recommande les tables managées pour les nouveaux workloads, car Unity Catalog peut gérer leur layout et leur maintenance. Choisir une table externe lorsque des données existantes doivent rester à un chemin précis ou être accédées directement par un système qui n'utilise pas les interfaces gouvernées.

Pour une table externe Delta éligible, Databricks recommande la conversion conservant le nom, l'historique, les permissions et les dépendances :

-- Nécessite Serverless ou Databricks Runtime 17.3 LTS ou supérieur.
ALTER TABLE production.sales.orders_external SET MANAGED;

SET MANAGED sans option s'applique à une table externe. Une table foreign éligible exige au contraire MOVE ou COPY :

-- Désactive la table du catalogue externe après la conversion.
ALTER TABLE foreign_catalog.sales.orders SET MANAGED MOVE;

-- Conserve la source externe et crée une copie managée distincte.
ALTER TABLE foreign_catalog.sales.orders SET MANAGED COPY;

Toutes les tables foreign ne sont pas éligibles : cette conversion vise les tables Delta fédérées via les intégrations prises en charge, notamment Hive metastore ou AWS Glue. La section 2.3.5 montre comment vérifier le type après l'opération.

La conversion externe se déroule en deux phases : copie initiale pendant laquelle lecteurs et écrivains continuent, puis courte bascule pendant laquelle les derniers commits sont transférés et les écritures brièvement bloquées. C'est pourquoi les deux avantages attendus dans le QCM sont : minimiser l'indisponibilité des lecteurs/écrivains et gérer les écritures concurrentes pendant la conversion.

Deux autres avantages à reconnaître :

• les lectures et écritures Databricks utilisant l'ancien chemin peuvent être redirigées vers la nouvelle table managée pour laisser fonctionner temporairement le code legacy compatible ; migrer malgré tout vers l'accès par nom catalog.schema.table ;
• la conversion d'une table externe peut être annulée avec UNSET MANAGED pendant la fenêtre de rollback.

Après la conversion, les streams de lecture ou d'écriture existants s'arrêtent volontairement avec DELTA_STREAMING_INTERRUPTED_BY_MANAGED_TABLE_CONVERSION. Il faut redémarrer le stream avec la même configuration et le même checkpoint : il reprend après le dernier offset/micro-batch validé et utilise la table convertie ou la redirection compatible. Ne pas supprimer le checkpoint, ce qui transformerait la reprise en nouvelle requête et pourrait retraiter des données. Vérifier aussi que le runtime prend en charge le mode d'accès utilisé, notamment pour un ancien accès par chemin.

Pendant la fenêtre de rollback d'une table externe convertie, actuellement 14 jours par défaut, UNSET MANAGED permet de revenir à l'emplacement externe d'origine :

ALTER TABLE production.sales.orders_external UNSET MANAGED;

La commande s'appelle bien UNSET MANAGED, et non SET UNMANAGED. C'est un rollback de conversion, pas une commande générale pour transformer n'importe quelle table managée en table externe. Après un foreign MOVE, il faut d'abord exécuter UNSET MANAGED, puis supprimer l'objet redevenu externe pour permettre sa refédération. Après un foreign COPY, la source n'ayant pas été désactivée, supprimer la copie managée suffit pour revenir à la table foreign lors de la synchronisation suivante. Respecter les procédures de rollback : supprimer directement une table issue d'un MOVE sans UNSET MANAGED peut entraîner une perte de données ou des incohérences.

Pour migrer une table managée ordinaire vers un stockage externe, créer explicitement une nouvelle table dans une external location, valider les données et les consommateurs, puis basculer. Ne pas déplacer manuellement les fichiers d'une table managée.

2.3.4 Copier avec CTAS, DEEP CLONE et SHALLOW CLONE

Un CTAS réalise une copie logique du résultat d'une requête. Comme la cible suivante ne déclare pas de LOCATION, elle est managée :

CREATE TABLE production.sales.orders_managed
AS SELECT * FROM production.sales.orders_external;

La nouvelle table contient les lignes et le schéma produits par le SELECT, mais elle ne reprend pas automatiquement l'historique Delta, le partitionnement, les propriétés, les contraintes, les tags ou les permissions propres à la source. CTAS permet en revanche de filtrer, joindre, agréger, renommer ou caster pendant la copie. La source et la cible sont ensuite indépendantes et aucune synchronisation continue n'est créée.

Un clone crée un nouvel objet ; il ne convertit pas la table source et ne conserve donc pas son identité Unity Catalog.

Mémo : CTAS copie un résultat logique et peut le transformer ; DEEP CLONE copie fidèlement l'état et la configuration d'une table, mais pas son historique ni ses droits Unity Catalog. Les deux créent une nouvelle table indépendante ; pour convertir l'objet existant sans changer son identité, utiliser SET MANAGED lorsqu'il est éligible.

CREATE TABLE dev.sandbox.orders_deep
DEEP CLONE production.sales.orders;

CREATE TABLE dev.sandbox.orders_shallow
SHALLOW CLONE production.sales.orders VERSION AS OF 12;

Version Python avec l'API Delta :

from delta.tables import DeltaTable

source = DeltaTable.forName(spark, "production.sales.orders")
source.clone(
    target="dev.sandbox.orders_deep",
    isShallow=False,
    replace=False,
)
source.cloneAtVersion(
    version=12,
    target="dev.sandbox.orders_shallow",
    isShallow=True,
    replace=False,
)

Chaque clone possède ensuite son propre historique. Modifier le clone ne modifie pas la source. Un shallow clone est rapide et peu coûteux, mais les fichiers référencés doivent rester disponibles ; pour les tables hors gestion UC, un VACUUM de la source peut notamment casser ces références. Un deep clone coûte plus de stockage et de temps à créer, mais il est autonome.

Pour convertir une table externe existante, préférer SET MANAGED à un deep clone :

Utiliser un deep clone lorsque l'objectif est réellement une seconde table autonome, pas pour changer le type de la table existante.

2.3.5 Inspecter avec DESCRIBE et information_schema

Les commandes ne répondent pas à la même question :

DESCRIBE CATALOG EXTENDED production;
DESCRIBE SCHEMA EXTENDED production.sales;

DESCRIBE TABLE production.sales.orders;
DESCRIBE TABLE EXTENDED production.sales.orders;
DESCRIBE TABLE EXTENDED production.sales.orders AS JSON;
DESCRIBE DETAIL production.sales.orders;

DATABASE est un ancien synonyme de SCHEMA, mais SCHEMA est préférable :

DESCRIBE DATABASE db_hr;
DESCRIBE SCHEMA EXTENDED db_hr;

Dans DESCRIBE TABLE EXTENDED, chercher la ligne Type pour distinguer MANAGED et EXTERNAL. Sur les runtimes récents, la variante AS JSON expose notamment les champs structurés row_filter et column_masks lorsqu'ils existent ; elle est préférable à l'analyse du rapport texte pour une automatisation. DESCRIBE SCHEMA EXTENDED décrit l'emplacement par défaut du schéma, pas le type de ses tables. DESCRIBE DETAIL renvoie une seule ligne structurée : son champ location fournit le chemin physique attribué et son champ format = 'delta' décrit le format, pas le statut managé ou externe.

Pour automatiser le contrôle de plusieurs objets, utiliser information_schema :

SELECT table_name, table_type
FROM system.information_schema.tables
WHERE table_catalog = 'production'
  AND table_schema = 'sales';

Une table étrangère constitue un cas particulier : certaines fédérations reproduisent le comportement du Hive metastore et DESCRIBE EXTENDED peut afficher EXTERNAL. Pour distinguer sûrement FOREIGN, utiliser Catalog Explorer ou system.information_schema.tables, dont table_type vaut FOREIGN. En Python, ces inspections SQL peuvent être exécutées avec spark.sql(...) ; il n'existe pas d'intérêt à dupliquer chaque DDL avec une API DataFrame.

2.3.6 Supprimer et récupérer avec DROP et UNDROP

L'effet de DROP TABLE dépend du propriétaire des fichiers :

Après le DROP d'une table managée Unity Catalog, les données ne disparaissent donc pas immédiatement. La table est récupérable avec UNDROP TABLE pendant 7 jours par défaut ; à l'expiration, Databricks supprime ensuite les fichiers sous-jacents du cloud dans un délai maximal annoncé de 48 heures.

DROP TABLE production.sales.orders;

-- Possible tant que la période de récupération n'est pas expirée.
UNDROP TABLE production.sales.orders;

-- Fonctionnalité actuelle en Public Preview : 0, ou une durée de 7 à 30 jours.
ALTER SCHEMA production.sales RETAIN DROPPED TO 14 DAYS;
DESCRIBE SCHEMA EXTENDED production.sales;

Le réglage du schéma prend le pas sur celui du catalogue et ne s'applique qu'aux tables supprimées après sa modification. Avec une rétention configurée à 0, la table n'est pas récupérable par UNDROP.

UNDROP TABLE restaure un objet managé supprimé. Ne pas le confondre avec RESTORE TABLE, qui crée une nouvelle version d'une table Delta existante à partir de son historique, ni avec le Time Travel, qui lit une ancienne version sans modifier l'état courant.

2.4 Delta Lake

Une table Delta est principalement constituée de :

• fichiers de données, généralement en Parquet ;
• journal de transactions _delta_log ;
• métadonnées et statistiques utilisées par le moteur.

Delta Lake ne remplace pas le stockage sous-jacent. C'est un format et une couche transactionnelle qui stocke ses fichiers de données et son journal dans un système durable, généralement du cloud object storage comme S3, ADLS ou GCS. Mémo de QCM : object storage + fichiers Parquet + _delta_log = table Delta.

Le journal fournit :

• transactions ACID ;
• concurrence optimiste entre lecteurs et écrivains ;
• contrôle et évolution du schéma ;
• DML (Data Manipulation Language, langage de manipulation des données) : INSERT, UPDATE, DELETE, MERGE ;
• historique, time travel et restauration ;
• traitement batch et streaming sur la même table.

Chaque commit Delta publie atomiquement une nouvelle version de la table. Un lecteur commencé sur une version continue de voir un snapshot cohérent, sans mélange partiel avec un commit concurrent : c'est la snapshot isolation, souvent expliquée comme un mécanisme MVCC (Multi-Version Concurrency Control). Les écritures utilisent en plus une concurrence optimiste ; un conflit incompatible peut faire échouer une transaction. Mémo : lecteur = snapshot versionné ; écrivain = nouveau commit atomique.

DESCRIBE DETAIL production.sales.orders;
DESCRIBE HISTORY production.sales.orders;

SELECT *
FROM production.sales.orders VERSION AS OF 12;

SELECT *
FROM production.sales.orders TIMESTAMP AS OF '2026-07-01T10:00:00Z';

RESTORE TABLE production.sales.orders TO VERSION AS OF 12;

Version PySpark avec l'API Delta :

from delta.tables import DeltaTable

orders = DeltaTable.forName(spark, "production.sales.orders")
orders.detail().show(truncate=False)
orders.history().show(truncate=False)

old_version_df = (
    spark.read.option("versionAsOf", 12)
         .table("production.sales.orders")
)
orders.restoreToVersion(12)

SQL est plus direct pour l'inspection interactive ; l'API PySpark est utile pour intégrer l'historique ou la restauration à un programme contrôlé.

À retenir : RESTORE ne rembobine pas le journal. Il crée une nouvelle version dont l'état correspond à l'ancienne. VACUUM peut supprimer les fichiers nécessaires aux anciennes versions et donc rendre certains time travels impossibles.

Un overwrite ou un CREATE OR REPLACE TABLE Delta crée lui aussi un nouveau commit : il remplace l'état courant, mais n'efface pas immédiatement l'historique. Tant que le journal et les anciens fichiers requis sont retenus, un time travel ou un RESTORE reste possible. VACUUM peut ensuite supprimer physiquement les fichiers devenus des tombstones et rendre ces versions illisibles.

2.5 Architecture médaillon

La couche Bronze n'est pas « mauvaise », la Silver n'est pas nécessairement agrégée et la Gold ne contient pas forcément une seule table. Chaque couche a un contrat et un public différents.

3. Ingestion et chargement

3.1 Choisir le mode d'ingestion

Toujours partir de la couche la plus managée qui satisfait le besoin, puis descendre vers une solution personnalisée si nécessaire.

3.2 Ingestion ponctuelle : Upload UI et CTAS

L'Upload UI est ponctuel et manuel. CTAS signifie Create Table As Select et crée la table à partir du résultat d'une requête.

CREATE OR REPLACE TABLE dev.bronze.orders  -- Incorrect : CREATE TABLE dev.silver.orders_typed (order_id BIGINT)
AS
SELECT
  *,
  CAST(order_id AS BIGINT) AS order_id,
  CAST(order_ts AS TIMESTAMP) AS order_ts,
  _metadata.file_name AS source_file,
  _metadata.file_modification_time AS source_modified_at,
  current_timestamp() AS ingested_at
FROM read_files(
  '/Volumes/dev/landing/raw/orders',
  format => 'csv',
  header => true,
  inferSchema => true
);

CTAS relit la source quand la commande est relancée. Ce n'est pas, à lui seul, un mécanisme d'ingestion incrémentale qui mémorise les fichiers déjà traités.

Avec CTAS, le SELECT détermine les noms et types des colonnes. On ne peut pas fournir simultanément une déclaration manuelle de colonnes et AS SELECT : Pour imposer un contrat séparément, créer d'abord la table avec son schéma, puis utiliser INSERT, COPY INTO ou MERGE. CREATE TABLE LIKE copie seulement la définition d'une table existante.

Équivalent PySpark :

from pyspark.sql import functions as F

orders_df = (
    spark.read
         .format("csv")
         .option("header", True)
         .option("inferSchema", True)
         .load("/Volumes/dev/landing/raw/orders")
         .withColumn("source_file", F.col("_metadata.file_name"))
         .withColumn("ingested_at", F.current_timestamp())
)

(orders_df.write
          .format("delta")
          .mode("overwrite")  # append
          .saveAsTable("dev.bronze.orders"))

mode("overwrite") remplace les données ; mode("append") ajoute des lignes. Le choix doit être explicite et compatible avec l'idempotence attendue.

Enregistrer directement des fichiers CSV comme table externe

CREATE TABLE dev.raw.my_table (
  col1 STRING,
  col2 STRING
)
USING CSV
OPTIONS (
  header = 'true',
  delimiter = ';'
)
LOCATION 's3://company-data/input/my_table';

Cette commande enregistre les fichiers existants à cet emplacement, avec le schéma et les options CSV indiqués. Elle crée une table externe CSV ; elle ne copie pas les données et ne les convertit pas en Delta. Il n'y a donc pas les mêmes garanties ACID, DML et time travel qu'avec une table Delta.

3.3 Ingestion incrémentale SQL : COPY INTO

COPY INTO charge dans une table Delta les fichiers qui n'ont pas déjà été enregistrés comme traités. La commande est relançable et idempotente : sans nouveau fichier, un nouveau run n'ajoute normalement rien.

CREATE TABLE IF NOT EXISTS dev.bronze.orders (
  order_id BIGINT,
  customer_id BIGINT,
  order_ts TIMESTAMP,
  amount DECIMAL(12, 2)
);

COPY INTO dev.bronze.orders
FROM '/Volumes/dev/landing/raw/orders'
FILEFORMAT = CSV  -- JSON
PATTERN = '*.csv' -- Glob facultatif : limite les fichiers éligibles par leur nom/chemin.
FORMAT_OPTIONS (
  'header' = 'true',
  'delimiter' = ',',
  'inferSchema' = 'true',
  'mode' = 'FAILFAST'
  -- , 'modifiedAfter' = '2026-08-01T00:00:00Z'
  )
COPY_OPTIONS (
  'mergeSchema' = 'true'
  -- , 'force' = 'true'
  );
-- VALIDATE ALL;
-- VALIDATE 100 ROWS;

mergeSchema=true : nouvelles colonnes peuvent être ajoutées à la cible.

PATTERN est un glob de sélection de fichiers, pas une condition sur leur contenu ; l'omettre charge tous les fichiers éligibles sous la source. Ici, mode=FAILFAST est une option du lecteur CSV placée dans FORMAT_OPTIONS : la première ligne mal formée fait échouer le chargement au lieu d'être tolérée ou ignorée.

Pour nommer quelques fichiers exacts plutôt qu'un glob, remplacer PATTERN par FILES = ('orders_001.csv', 'archive/orders_002.csv'). La liste est limitée à 1 000 fichiers ; les noms sont relatifs au chemin de FROM. Utiliser FILES ou PATTERN, pas les deux.

modifiedAfter est une option de lecture du format, d'où sa présence dans FORMAT_OPTIONS. Elle filtre les fichiers sur leur métadonnée de modification ; elle ne filtre pas une colonne métier contenue dans le JSON. L'historique de COPY INTO continue ensuite à empêcher le rechargement des fichiers déjà traités, sauf force=true.

force=true désactive l'idempotence de COPY INTO et recharge les fichiers éligibles même s'ils figurent déjà dans l'historique de chargement. L'utiliser pour un replay explicitement contrôlé, avec une cible vidée/dédupliquée ou une stratégie d'upsert ; sinon il crée facilement des doublons.

VALIDATE permet de vérifier les fichiers et le schéma sans écrire dans la table cible : ALL valide toutes les lignes éligibles ; n ROWS limite le résultat de validation. C'est utile avant un premier chargement ou après un changement de format. Il s'agit d'une clause SQL de COPY INTO, sans équivalent propre dans DataFrameWriter.

À retenir :

• la destination doit être une table Delta ;
• les sources usuelles sont S3, ADLS, GCS ou un volume Unity Catalog ;
• COPY INTO convient à une ingestion incrémentale simple ;
• le pattern sûr suppose des fichiers source immuables et un nom unique par livraison : un fichier déjà chargé est ignoré lors des runs suivants, même si son contenu a été modifié ; seul force=true demande explicitement son rechargement ;
• pour des millions de fichiers, une faible latence ou un schéma complexe, préférer Auto Loader.

Les privilèges Unity Catalog nécessaires dépendent notamment d'un accès par external location ou par volume ; ils sont regroupés avec le modèle de droits en 8.2.

COPY INTO est une commande SQL sans équivalent natif dans l'API DataFrame. Une lecture puis un append PySpark n'aurait pas le même suivi des fichiers déjà ingérés ; utiliser au besoin spark.sql("COPY INTO ..."), sans considérer cela comme une seconde implémentation.

Si les chargements créent beaucoup de petits fichiers, optimized writes et auto compaction peuvent en limiter la création côté table Delta ; ce ne sont pas des COPY_OPTIONS. Pour compacter des petits fichiers déjà écrits, lancer OPTIMIZE comme en 7.5.3.

3.4 Lakeflow Connect : connecteurs standards

Un connecteur standard identifie la source et ses capacités ; Structured Streaming est l'un des moteurs permettant de l'utiliser. Selon le niveau de contrôle recherché :

Avec Structured Streaming, une source incrémentale est généralement traitée sous forme de micro-batches successifs et sa progression est conservée dans un checkpoint. Deux sources standards sont particulièrement importantes pour l'examen :

Le trigger détermine quand les micro-batches sont lancés ; le sink reçoit leur résultat ; le checkpoint permet la reprise après une panne. Les transformations stateful, watermarks, output modes, foreachBatch et streaming joins sont détaillés dans la partie 4.

3.4.1 Auto Loader : ingestion incrémentale de fichiers

Auto Loader est la source Structured Streaming cloudFiles. Il découvre progressivement les fichiers et s'appuie sur un checkpoint pour mémoriser leur traitement.

Lecture rapide : le minimum commun est source Auto Loader + format des fichiers + chemin source + sink. Pour lire du JSON/CSV/XML avec Auto Loader PySpark autonome, ajouter aussi un schéma ou schemaLocation ; pour une exécution durable et reprenable, ajouter le checkpoint.

Version SQL dans un pipeline Lakeflow : read_files sous STREAM utilise Auto Loader ; Lakeflow gère schemaLocation, le checkpoint, le trigger et l'écriture dans la cible.

-- Sink (destination) de la requête
CREATE OR REFRESH STREAMING TABLE dev.bronze.orders
AS
SELECT *, _metadata.file_name AS source_file
FROM STREAM read_files(
  -- Source Auto Loader et format réel des fichiers
  '/Volumes/dev/landing/raw/orders',
  format => 'json',

  -- Deviner schéma selon échantillon
  inferColumnTypes => true,
  -- Indiquer schéma partiellement
  schemaHints => 'event_id BIGINT, customer.address.zip STRING, items.element.price DECIMAL(12,2), _corrupt_record STRING',
  -- Expliciter schéma (pas inferColumnTypes, schemaHints, addNewColumns) : .schema("event_id BIGINT, event_ts TIMESTAMP, payload STRING")
  -- schema => 'event_id BIGINT, event_ts TIMESTAMP, payload STRING',

  -- Si nouvelle colonne : addNewColumns, failOnNewColumns, rescue ou none
  schemaEvolutionMode => 'addNewColumns',
  -- Si schemaEvolutionMode=rescue
  rescuedDataColumn => '_rescued_data',

  -- Inclure les fichier existants (valable la 1ere fois)
  includeExistingFiles => true,

  -- Si lignes d'un json/csv mal formées : PERMISSIVE (isoler), DROPMALFORMED (supprime ligne), FAILFAST (arrête lecture)
  -- Si PERMISSIVE (mettre schemaHints : _corrupt_record STRING)
  mode => 'PERMISSIVE',
  columnNameOfCorruptRecord => '_corrupt_record',

  -- Reprendre des fichiers source modifiés
  allowOverwrites => true,

  -- Pour File notification/file events
  useManagedFileEvents => true,

  -- Si problème de débit/mémoire/durée des batchs 
  maxFilesPerTrigger => 500,
  maxBytesPerTrigger => '10g'
);

-- Alternative contrôlée hors gestion automatique Lakeflow :
-- ALTER TABLE dev.bronze.orders ADD COLUMNS (phone STRING);

Version PySpark Structured Streaming autonome :

raw_df = (
    spark.readStream
         # Source Auto Loader et format réel des fichiers
         .format("cloudFiles")
         .option("cloudFiles.format", "json")
         .option("cloudFiles.schemaLocation", "/Volumes/dev/checkpoints/orders_schema")

         # Deviner schéma selon échantillon
         .option("cloudFiles.inferColumnTypes", "true")
         # Indiquer schéma partiellement
         .option(
             "cloudFiles.schemaHints",
             "event_id BIGINT, customer.address.zip STRING, "
             "items.element.price DECIMAL(12,2), _corrupt_record STRING",
         )
         # Expliciter schéma (pas inferColumnTypes, schemaHints, addNewColumns) : .schema("event_id BIGINT, event_ts TIMESTAMP, payload STRING")

         # Si nouvelle colonne : addNewColumns, failOnNewColumns, rescue ou none
         .option("cloudFiles.schemaEvolutionMode", "addNewColumns")
         ## Si schemaEvolutionMode=rescue
         .option("rescuedDataColumn", "_rescued_data")

         # Inclure les fichier existants (valable la 1ere fois)
         .option("cloudFiles.includeExistingFiles", "true")

         # Si lignes d'un json/csv mal formées : PERMISSIVE (isoler), DROPMALFORMED (supprime ligne), FAILFAST (arrête lecture)
         ## Si PERMISSIVE (mettre schemaHints : _corrupt_record STRING)
         .option("mode", "PERMISSIVE")
         .option("columnNameOfCorruptRecord", "_corrupt_record")

         # Reprendre des fichiers source modifiés
         .option("cloudFiles.allowOverwrites", "true")

         # Pour File notification/file events
         .option("cloudFiles.useManagedFileEvents", "true")

         # Si problème de débit/mémoire/durée des batchs 
         .option("cloudFiles.maxFilesPerTrigger", 500)
         .option("cloudFiles.maxBytesPerTrigger", "10g")

         # Chemin source (placé à la fin par la syntaxe DataFrame)
         .load("/Volumes/dev/landing/raw/orders")
)

query = (
    raw_df.writeStream
          # Chemin du checkpoint
          .option("checkpointLocation", "/Volumes/dev/checkpoints/orders_data")
          # Exécution : AvailableNow ou processingTime="1 minute"
          .trigger(availableNow=True)
          # Si la table Delta doit évoluer
          .option("mergeSchema", "true")
          # Sink (destination) de la requête
          .toTable("dev.bronze.orders")
)

1. Définir la source et son format

.format("cloudFiles") sélectionne Auto Loader, cloudFiles.format indique le format réel et .load(...) fournit le chemin source. Les formats usuels sont JSON, CSV, XML, Parquet, Avro, ORC, texte et binaryFile. En SQL Lakeflow, les équivalents sont STREAM read_files(path, format => ...) et CREATE OR REFRESH STREAMING TABLE.

2. Construire et conserver le schéma lu

schemaLocation conserve le schéma découvert et son historique d'évolution, pas la progression des fichiers. Il doit être durable et propre à la lecture ; Lakeflow en gère automatiquement l'emplacement.

Pour construire le schéma du DataFrame source, choisir une stratégie cohérente :

• un schema complet constitue un contrat strict et remplace l'inférence ;
• sans schéma explicite, Auto Loader PySpark infère par défaut les colonnes JSON, CSV et XML en STRING ; cloudFiles.inferColumnTypes=true demande une inférence typée sur un échantillon ;
• schemaHints impose seulement les types critiques, y compris dans une structure imbriquée (customer.address.zip) ou un élément d'array (items.element.price). Un hint guide la lecture ; ce n'est pas un CAST métier.

L'inférence reste un échantillonnage, pas un contrat. Parquet porte déjà ses types. Lorsqu'un champ connu du schéma unifié manque dans un document, sa valeur est NULL pour cette ligne.

3. Choisir la politique d'évolution et les rescued data

Choisir un seul cloudFiles.schemaEvolutionMode :

Avec addNewColumns, les anciennes lignes valent NULL pour la nouvelle colonne ; le mode n'élargit pas arbitrairement les types existants. _rescued_data peut aussi conserver des champs hors schéma, des différences de casse ou des valeurs incompatibles avec le type attendu. Une incompatibilité imbriquée se traite avec un schéma explicite, des schemaHints, les rescued data ou une normalisation Silver.

4. Décider du chargement des fichiers déjà présents

cloudFiles.includeExistingFiles n'est évalué qu'au premier démarrage associé au checkpoint : true charge le backlog puis les nouveaux fichiers ; false ne traite que les fichiers arrivant après l'initialisation. Modifier cette option en conservant le checkpoint n'a aucun effet.

5. Capturer les documents mal formés

mode concerne les documents JSON ou CSV réellement mal formés :

Pour utiliser columnNameOfCorruptRecord='_corrupt_record', cette colonne doit être déclarée en STRING, comme dans les schemaHints de l'exemple. Ne pas la confondre avec _rescued_data : la première vise une syntaxe illisible, la seconde des données lisibles mais hors schéma ou hors type. badRecordsPath peut isoler les erreurs dans un répertoire, mais une colonne de corruption conserve plus simplement le lien avec la ligne ingérée.

6. Décider du retraitement des fichiers écrasés

Auto Loader suppose normalement des fichiers immuables et cloudFiles.allowOverwrites vaut false par défaut. Avec true, un fichier modifié au même chemin peut être relu en entier. Un sink append ne supprime pas son ancienne version : il peut donc produire des doublons et nécessite une déduplication ou un MERGE idempotent.

7. Choisir le mode de découverte des fichiers

cloudFiles.useManagedFileEvents=true utilise le service de file events mutualisé d'une external location sur laquelle ces événements ont été activés. cloudFiles.useNotifications=true choisit au contraire le mode classique, avec une infrastructure de notification propre au stream ; ne pas combiner les deux options.

Le mode classique nécessite les permissions cloud permettant de créer, configurer et consommer les ressources d'événements : SNS/SQS sur AWS, Event Grid/Queue Storage sur Azure ou topics/subscriptions Pub/Sub sur GCP. Sur AWS, le chemin est événement S3 → SNS → SQS → Auto Loader : SNS (Simple Notification Service) diffuse l'événement et SQS (Simple Queue Service) le conserve jusqu'à sa consommation. La notification contient les métadonnées du fichier, pas son contenu.

Si des fichiers manquent, vérifier les préfixes et suffixes configurés, SNS/SQS, IAM et la région. Un retry SQS ne doit pas, à lui seul, dupliquer les lignes avec un checkpoint stable ; contrôler plutôt le checkpoint, allowOverwrites, l'immuabilité de la source et l'idempotence du sink.

8. Régler l'admission de chaque micro-batch

Ces options sont secondaires et se règlent après observation d'un problème de débit, de mémoire ou de durée :

• maxFilesPerTrigger est une borne dure sur le nombre de fichiers ;
• maxBytesPerTrigger est une borne souple sur leur volume et peut être dépassée pour admettre au moins un fichier ;
• si les deux sont définies, le premier seuil atteint borne le batch.

Sur une streaming table exécutée par un SQL warehouse serverless, laisser de préférence l'admission dynamique dimensionner les batches.

9. Conserver la progression dans le checkpoint

checkpointLocation répond à « jusqu'où la requête est-elle arrivée ? ». Il conserve les fichiers ou offsets lus, les commits validés dans le sink et l'état des opérations stateful. Après une panne, Spark reprend après le dernier micro-batch validé.

• Chaque requête doit avoir son propre checkpoint durable.
• Le supprimer ou changer son chemin crée une nouvelle requête, avec un risque de retraitement et de doublons.
• Changer de source, de sink ou la structure d'une opération stateful exige généralement un nouveau checkpoint ; modifier un filtre, un trigger ou une limite d'admission est généralement compatible.
• Un checkpoint stable et un sink Delta transactionnel permettent une application exactly-once des fichiers découverts. foreachBatch reste at-least-once par défaut et doit être idempotent.

Ne pas confondre les deux états persistés :

Ils peuvent partager une racine, mais doivent utiliser des sous-répertoires distincts. Des doublons après une évolution ne se corrigent donc pas en supprimant schemaLocation : vérifier d'abord le checkpoint, allowOverwrites et l'idempotence du sink.

10. Choisir le trigger PySpark

Une requête utilise au plus un trigger explicite :

• .trigger(availableNow=True) draine le backlog présent au démarrage en autant de micro-batches que nécessaire, en respectant les limites d'admission, puis s'arrête. Les fichiers arrivés ensuite attendent le prochain run ;
• .trigger(processingTime="1 minute") laisse la requête active et tente de lancer un micro-batch chaque minute. Si un batch dure davantage, le suivant démarre dès que possible ;
• sans .trigger(...), Spark lance par défaut les micro-batches aussi vite que possible.

Ces appels concernent un writeStream autonome. En SQL Lakeflow, choisir un pipeline triggered ou continuous, puis configurer sa planification dans le pipeline ou Lakeflow Jobs.

11. Faire évoluer le schéma cible et écrire dans le sink

Auto Loader et la table Delta cible gèrent deux schémas indépendants :

schema / schemaHints / schemaEvolutionMode
        → schéma lu par Auto Loader et produit dans le DataFrame

ALTER TABLE / mergeSchema
        → schéma de la table Delta cible à l'écriture

Delta applique le schema enforcement à l'écriture. Une colonne inconnue ou un type non convertible fait échouer l'opération. Si Auto Loader ajoute une colonne, préparer la cible avec ALTER TABLE ou autoriser son évolution avec mergeSchema=true. Omettre mergeSchema lorsque le contrat doit rester strict ; il est inutile si les schémas sont déjà compatibles ou si Lakeflow gère directement sa streaming table. Il ne corrige jamais une erreur de parsing en amont.

Le sink est la destination de la requête. Dans l'exemple, .toTable("dev.bronze.orders") écrit dans une table Delta. Choisir SQL Lakeflow pour une ingestion déclarative ; choisir Structured Streaming autonome lorsqu'il faut contrôler directement le checkpoint, le trigger ou une logique programmatique. Dans les deux cas, Auto Loader ne garantit pas l'ordre de découverte des fichiers.

3.4.2 Fichiers non structurés avec binaryFile

Pour les fichiers non structurés, binaryFile crée une ligne par fichier avec notamment son chemin, sa date de modification, sa longueur et son contenu binaire. C'est utile pour gouverner des images ou documents avant extraction.

Version SQL dans un pipeline :

CREATE OR REFRESH STREAMING TABLE dev.bronze.invoice_images AS
SELECT *
FROM STREAM read_files(
  's3://shop/raw/invoices/',
  format => 'binaryFile',
  pathGlobFilter => '*.jpg'
);

Version PySpark :

images_df = (
    spark.readStream
         .format("cloudFiles")
         .option("cloudFiles.format", "binaryFile")
         .option("pathGlobFilter", "*.jpg")
         .load("s3://shop/raw/invoices/")
)

pathGlobFilter sélectionne les noms de fichiers sans parcourir le contenu. Conserver la colonne content seulement si elle est réellement nécessaire, car elle peut être volumineuse.

3.4.3 Kafka : ingestion événementielle

Kafka est une source événementielle lue avec Structured Streaming. Contrairement à Auto Loader, qui découvre des fichiers, Kafka fournit des messages ordonnés à l'intérieur de chaque partition et suivis par offsets.

Version SQL dans un pipeline :

CREATE OR REFRESH STREAMING TABLE dev.bronze.kafka_events AS
SELECT *
FROM STREAM read_kafka(
  bootstrapServers => 'host:port',
  subscribe => 'events_topic',
  startingOffsets => 'latest'
);

Version PySpark :

events_stream = (
    spark.readStream
         .format("kafka")
         .option("kafka.bootstrap.servers", "host:port")
         .option("subscribe", "events_topic")
         .option("startingOffsets", "latest")
         .load()
)

La source contient notamment key et value en BINARY, ainsi que topic, partition, offset et timestamp. Il faut généralement décoder value, puis appliquer from_json ou parse_json lors de la transformation Bronze vers Silver. startingOffsets = latest ne lit que les nouveaux messages lors du premier démarrage ; après création du checkpoint, un redémarrage reprend les offsets enregistrés et ignore cette option initiale. Utiliser exactement une stratégie parmi subscribe, subscribePattern et assign.

Pour un topic à fort débit — par exemple 10 000 événements/s — avec une latence attendue inférieure à une minute, choisir Structured Streaming de Kafka vers Delta, avec un trigger adapté et un checkpoint durable. Une lecture batch JDBC, COPY INTO ou un Job quotidien ne répond pas à ce besoin. Dimensionner ensuite les partitions Kafka et le compute d'après le débit réel ; le seul choix de l'API ne garantit pas le SLA.

3.5 Lakeflow Connect : connecteurs managés

Un connecteur managé crée un pipeline d'ingestion depuis l'UI, une API ou un bundle pour une source prise en charge. Databricks gère davantage :

• l'authentification et les particularités de la source ;
• la capture incrémentale ou le CDC lorsqu'il est proposé ;
• les retries, la reprise et le cycle de vie du pipeline ;
• l'évolution de schéma prise en charge ;
• la maintenance des API, les cas limites et le monitoring.

Au premier run, le connecteur effectue généralement un snapshot initial des objets sélectionnés, puis charge les changements de façon incrémentale lorsque la source et l'objet le permettent. La technique exacte dépend du connecteur : CDC/change tracking pour certaines bases, ou curseur/high-water mark pour certaines API. Il ne faut donc pas généraliser « tous les connecteurs font du CDC ».

Pour un connecteur SaaS/API pris en charge, Databricks encapsule notamment la pagination, les limites d'appel, les retries et les évolutions de schéma que ce connecteur sait gérer. Ce bénéfice ne s'applique pas automatiquement à une API REST personnalisée ni de façon identique à toutes les sources.

Les connecteurs managés couvrent notamment des applications SaaS et des bases d'entreprise. Il existe aussi des connecteurs de streaming managés : par exemple, le connecteur Kafka managé, actuellement en bêta, utilise une CONNECTION Unity Catalog et écrit dans des streaming tables.

Le connecteur PostgreSQL managé peut exploiter la réplication logique et le WAL pour capturer les changements en CDC et les appliquer à des streaming tables, lorsqu'il est disponible dans le workspace.

Lakeflow Connect ou Lakehouse Federation

• Lakeflow Connect copie les données dans des tables Delta Unity Catalog afin de les historiser, transformer et servir localement.
• Lakehouse Federation interroge la source distante sans l'ingérer. Une CONNECTION Unity Catalog conserve le type de source, l'hôte et les informations d'authentification ; un foreign catalog expose ensuite ses objets dans le namespace à trois niveaux.

La création de la CONNECTION, du FOREIGN CATALOG et l'exposition des tables étrangères sont détaillées en 2.3.2. Mémo : Connect ingère une copie locale ; Federation interroge la source sans la déplacer.

3.6 Sources personnalisées : JDBC, ODBC et REST

Spark peut lire une base relationnelle en JDBC :

jdbc_df = (
    spark.read
         .format("jdbc")
         .option("url", "jdbc:postgresql://host:5432/app")
         .option("dbtable", "public.orders")
         .option("user", dbutils.secrets.get("ingestion", "user"))
         .option("password", dbutils.secrets.get("ingestion", "password"))
         .option("driver", "org.postgresql.Driver")
         .load()
)

jdbc_df.write.mode("append").saveAsTable("dev.bronze.orders_jdbc")

Le nom passé à option("driver", ...) désigne une classe qui doit être disponible sur le classpath. Si le runtime ne fournit pas le pilote, l'ajouter à la task Lakeflow Job dans Dependent libraries — de préférence par coordonnées Maven, ou comme JAR — afin qu'il soit installé avant l'exécution. Avec un compute standard, le chemin ou les coordonnées peuvent devoir figurer dans l'allowlist ; sur serverless, configurer Environment and Libraries ou utiliser une connexion JDBC Unity Catalog lorsqu'elle est prise en charge. Une ClassNotFoundException sur le driver indique d'abord une dépendance absente, pas une panne réseau.

Le connecteur Spark JDBC standard est une source batch : il utilise spark.read, pas spark.readStream. Déclarer sa sortie comme streaming table ne rend pas la source JDBC continue. Dans un pipeline déclaratif, une lecture JDBC standard alimente donc plutôt une materialized view ou une étape batch planifiée.

Une erreur connection timeout signifie qu'aucune connexion n'a été établie dans le délai ; elle oriente d'abord vers le réseau ou une adresse incorrecte, pas vers le mot de passe. Vérifier le host, le port, le DNS, les routes, le peering/VPN et les security groups/NSG/firewalls autorisant le trafic depuis le compute ou la passerelle Databricks vers MySQL, généralement sur le port 3306. Selon l'architecture, il faut autoriser les plages IP adaptées ou mettre en place une connectivité privée ; « ajouter l'IP du workspace » n'est donc pas une règle universelle. Une erreur Access denied relève au contraire des credentials ou privilèges MySQL.

Une ingestion JDBC personnalisée peut devenir incrémentale par runs batch en conservant un curseur, par exemple updated_at, puis en ne demandant que les lignes postérieures au dernier curseur validé. La requête exécutée par la base ressemble à :

SELECT order_id, amount, updated_at
FROM public.orders
WHERE updated_at > TIMESTAMP '2026-08-16 10:00:00'
ORDER BY updated_at, order_id;

last_cursor = "2026-08-16 10:00:00"
incremental_query = f"""
  SELECT order_id, amount, updated_at
  FROM public.orders
  WHERE updated_at > TIMESTAMP '{last_cursor}'
"""

changes_df = (
    spark.read.format("jdbc")
         .option("url", "jdbc:postgresql://host:5432/app")
         .option("query", incremental_query)
         .option("user", dbutils.secrets.get("ingestion", "user"))
         .option("password", dbutils.secrets.get("ingestion", "password"))
         .load()
)

Persister le nouveau curseur seulement après une écriture cible réussie et idempotente. Un timestamp seul peut avoir des ex æquo : utiliser au besoin un curseur composite (updated_at, order_id), prévoir un petit chevauchement puis un MERGE, et traiter explicitement les suppressions. Lorsqu'un connecteur Lakeflow managé avec CDC ou cursor_column existe, il est préférable à ce polling artisanal.

Si la source ne possède aucune colonne incrémentale fiable, le fallback est un snapshot batch complet vers une table de staging, une déduplication déterministe par clé, puis un MERGE dans la cible. Cette solution est correcte mais relit toute la source ; elle doit donc rester un compromis. Les suppressions nécessitent en plus une information CDC ou, avec beaucoup de prudence, une synchronisation par WHEN NOT MATCHED BY SOURCE THEN DELETE :

MERGE INTO dev.silver.orders AS target
USING dev.staging.orders_snapshot_deduplicated AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
WHEN NOT MATCHED BY SOURCE THEN DELETE;

En PySpark, lire le snapshot avec spark.read.format("jdbc"), le dédupliquer avec une fenêtre comme en 4.1.4, puis appliquer le même MERGE via DeltaTable. Ne jamais ajouter la clause de suppression si le snapshot peut être partiel.

Sans options de partitionnement, une lecture JDBC Spark utilise une seule partition et une seule connexion : elle peut devenir le goulot d'étranglement d'une grosse extraction. Pour paralléliser, fournir ensemble une colonne numérique/date/timestamp et ses bornes (partitionColumn, lowerBound, upperBound, numPartitions). Les bornes calculent les plages mais ne filtrent pas les lignes ; numPartitions borne aussi le nombre de connexions JDBC concurrentes. Trop de partitions peuvent surcharger la base source.

ODBC est surtout utilisé par des applications clientes et outils BI pour se connecter à un SQL warehouse. Un client Python ODBC peut aussi extraire une petite source et convertir les résultats en DataFrame, mais JDBC distribué ou un connecteur Lakeflow est généralement préférable pour les gros volumes.

Pour une API REST propriétaire sans connecteur et un petit payload quotidien — par exemple moins de 1 Mo — un script Python exécuté dans un notebook ou une task Job est pragmatique. L'appel HTTP se fait sur le driver : valider le statut et le schéma, puis écrire la réponse brute ou normalisée en Bronze.

import requests
from pyspark.sql import types as T

orders_schema = T.StructType([
    T.StructField("order_id", T.LongType(), False),
    # Bronze : type simple et robuste ; convertir en DECIMAL dans Silver si nécessaire.
    T.StructField("amount", T.DoubleType(), True),
])

token = dbutils.secrets.get("ingestion", "crm_api_token")
response = requests.get(
    "https://api.example.com/orders",
    headers={"Authorization": f"Bearer {token}"},
    timeout=30,
)
response.raise_for_status()

orders_df = spark.createDataFrame(response.json()["orders"], orders_schema)
(orders_df.write
          .format("delta")
          .mode("append")
          .saveAsTable("dev.bronze.orders_api"))

Pour la production, gérer pagination, rate limits, backoff exponentiel, validation et idempotence dans Lakeflow Jobs. Ne jamais mettre le token en clair dans le notebook, Git ou les logs. Si les volumes deviennent importants, l'API doit être extraite par pages vers du stockage objet plutôt que parallélisée aveuglément depuis les executors.

4. Transformation et modélisation

4.1 De Bronze à Silver

L'objectif est de transformer une donnée brute et traçable en donnée fiable, typée et réutilisable. Les fondamentaux Spark ci-dessous s'appliquent aussi aux autres couches ; ils sont présentés ici car Bronze → Silver constitue généralement la première transformation substantielle.

4.1.1 Fondamentaux Spark et choix de l'interface

• Un DataFrame est distribué et immuable : une transformation retourne un nouveau DataFrame.
• Les transformations sont généralement lazy. Elles construisent un plan ; une action comme count, collect, write ou display déclenche l'exécution.
• Les fonctions Spark natives sont préférables aux UDF Python : elles sont visibles par l'optimiseur et évitent souvent des coûts de sérialisation.
• collect() est une action qui ramène toutes les lignes du DataFrame sous forme d'objets Python dans la mémoire du driver. toPandas() fait de même dans un DataFrame pandas. Les réserver à de petits résultats explicitement bornés, sinon le driver peut subir un OOM.
• Un shuffle est une redistribution des données entre partitions et executors, généralement par clé. Il implique réseau, sérialisation et parfois disque ; groupBy, distinct, les tris et la plupart des joins peuvent en provoquer un.

Imports usuels :

from pyspark.sql import functions as F
from pyspark.sql import types as T
from pyspark.sql.window import Window

Raccourci de choix : SQL est généralement le plus lisible pour les transformations déclaratives, agrégations, vues et objets de pipeline ; PySpark est préférable pour modulariser en fonctions, tester, utiliser boucles/conditions ou des bibliothèques Python. Les deux utilisent le moteur Spark et peuvent coexister dans un pipeline, dans des fichiers distincts.

Fonctions natives et UDF

Une UDF (User-Defined Function, fonction définie par l'utilisateur) encapsule une logique réutilisable lorsque les fonctions intégrées ne suffisent pas. Une fonction SQL Unity Catalog peut être gouvernée et appelée comme une fonction native :

CREATE OR REPLACE FUNCTION dev.utils.plus_one(value INTEGER)
RETURNS INTEGER
RETURN value + 1;

SELECT dev.utils.plus_one(5);  -- 6

Les opérations arithmétiques propagent NULL : plus_one(NULL) retourne donc NULL. Préférer une expression native telle que value + 1 lorsqu'elle n'a pas besoin d'être centralisée, car les fonctions natives sont plus simples à optimiser. En PySpark, le même calcul ne nécessite pas de Python UDF :

result_df = source_df.withColumn("value_plus_one", F.col("value") + 1)

Pour exécuter une fonction Unity Catalog, un principal a besoin de USE CATALOG, USE SCHEMA et EXECUTE sur la fonction.

coalesce() ou repartition()

Version SQL avec des hints de partitionnement :

SELECT /*+ COALESCE(20) */ *
FROM dev.silver.orders;

SELECT /*+ REPARTITION(80, customer_id) */ *
FROM dev.silver.orders;

Version PySpark :

# 200 -> 20 partitions, peu coûteux mais potentiellement déséquilibré.
fewer_df = filtered_df.coalesce(20)

# Shuffle vers 80 partitions réparties selon customer_id.
balanced_df = filtered_df.repartition(80, "customer_id")

Attention au vocabulaire : df.coalesce(20) agit sur les partitions ; la fonction SQL coalesce(country, 'UNKNOWN') retourne la première valeur non NULL. Même nom, objectif différent. Éviter aussi coalesce(1)/repartition(1) sur un gros volume : une seule task devient le goulot d'étranglement.

4.1.2 Nettoyage, typage et normalisation

Nettoyer Bronze vers Silver

Version SQL :

CREATE OR REPLACE TABLE dev.silver.orders AS
SELECT
  CAST(order_id AS BIGINT) AS order_id,
  to_timestamp(order_ts) AS order_ts,
  CAST(amount AS DECIMAL(12, 2)) AS amount,
  upper(trim(coalesce(country, 'UNKNOWN'))) AS country,
  CASE
    WHEN amount >= 1000 THEN 'HIGH'
    WHEN amount >= 100 THEN 'MEDIUM'
    ELSE 'LOW'
  END AS amount_band,
  customer AS customer_id
FROM dev.bronze.orders
WHERE order_id IS NOT NULL;

Version PySpark :

silver_df = (
    spark.table("dev.bronze.orders")
         .filter(F.col("order_id").isNotNull())
         .withColumn("order_id", F.col("order_id").cast("long"))
         .withColumn("order_ts", F.to_timestamp("order_ts"))
         .withColumn("amount", F.col("amount").cast("decimal(12,2)"))
         .withColumn("country", F.upper(F.trim("country")))
         .withColumn(
             "amount_band",
             F.when(F.col("amount") >= 1000, F.lit("HIGH"))
              .when(F.col("amount") >= 100, F.lit("MEDIUM"))
              .otherwise(F.lit("LOW"))
         )
         .drop("temporary_column")
         .withColumnRenamed("customer", "customer_id")
)

(silver_df.write
          .format("delta")
          .mode("overwrite")
          .saveAsTable("dev.silver.orders"))

Pour normaliser un e-mail, appliquer TRIM avant LOWER ; les deux fonctions conservent NULL :

SELECT
  user_id,
  lower(trim(email)) AS email
FROM dev.bronze.users;

from pyspark.sql import functions as F

users_df = spark.table("dev.bronze.users")
clean_users_df = users_df.withColumn(
    "email",
    F.lower(F.trim(F.col("email"))),
)

Pour construire une chaîne avec un séparateur, concat_ws ignore automatiquement les arguments NULL. Il évite donc un séparateur inutile lorsque, par exemple, le deuxième prénom est absent ; une chaîne vide reste toutefois différente de NULL.

SELECT concat_ws(
         ' ',
         trim(first_name),
         trim(middle_name),
         trim(last_name)
       ) AS full_name
FROM dev.bronze.users;

users_df = users_df.withColumn(
    "full_name",
    F.concat_ws(
        " ",
        F.trim("first_name"),
        F.trim("middle_name"),
        F.trim("last_name"),
    ),
)

Pour remplacer ou supprimer des caractères individuels, translate évite une expression régulière : chaque caractère de la deuxième chaîne correspond à celui de même position dans la troisième ; si cette dernière est plus courte, les caractères sans correspondance sont supprimés.

SELECT translate(phone, ' ()-+', '') AS phone_digits
FROM dev.bronze.customers;

customers_df = customers_df.withColumn(
    "phone_digits",
    F.translate("phone", " ()-+", ""),
)

Utiliser plutôt replace pour une sous-chaîne exacte et regexp_replace pour un motif.

Dans l'exemple de nettoyage précédent, withColumn ajoute une colonne ou remplace celle qui porte déjà le même nom ; when(...).otherwise(...) est l'équivalent DataFrame de CASE WHEN ... ELSE ... END. Pour renommer toutes les colonnes d'un coup, toDF applique les nouveaux noms par position et exige exactement autant de noms que le DataFrame possède de colonnes :

SELECT *, 'CRM_PROD' AS source_system
FROM dev.bronze.customers;

tagged_df = customers_df.withColumn(
    "source_system",
    F.lit("CRM_PROD"),
)

F.lit("CRM_PROD") construit une valeur constante pour chaque ligne ; F.col("CRM_PROD") chercherait au contraire une colonne portant ce nom.

SELECT
  id AS order_id,
  customer AS customer_id,
  total AS amount
FROM dev.bronze.orders;

renamed_df = source_df.toDF("order_id", "customer_id", "amount")

Pour une seule colonne, préférer withColumnRenamed ou un alias dans select, plus explicite et moins fragile si l'ordre du schéma change.

selectExpr permet d'écrire des expressions SQL sous forme de chaînes dans une transformation DataFrame. Il est pratique pour une projection courte avec casts et alias :

SELECT CAST(reading_value AS DOUBLE) AS sensor_value
FROM dev.bronze.sensors;

sensors_df = (
    spark.read.table("dev.bronze.sensors")
         .selectExpr("CAST(reading_value AS DOUBLE) AS sensor_value")
)

Comme SELECT, selectExpr ne conserve que les colonnes énumérées. Utiliser plutôt .select(F.col("reading_value").cast("double").alias("sensor_value")) lorsque l'on souhaite une API PySpark typée, et TRY_CAST/try_cast si une valeur mal formée doit devenir NULL au lieu de faire échouer le traitement.

Gestion des valeurs manquantes :

# Supprime les lignes sans identifiant ; remplit uniquement certaines colonnes.
clean_df = (
    source_df.na.drop(subset=["order_id"])
             .na.fill({"country": "UNKNOWN", "quantity": 0})
)

# Autre règle possible : conserver les lignes ayant au moins 8 valeurs définies.
complete_enough_df = source_df.na.drop(thresh=8)

dropna et na.drop sont des alias. dropna(thresh=8) conserve uniquement les lignes qui possèdent au moins huit valeurs non nulles ; thresh prend le pas sur how. Le paramètre subset=[...] limite les colonnes examinées. En SQL, il n'existe pas de méthode dropna : écrire explicitement le prédicat WHERE correspondant. NULL n'est ni zéro ni chaîne vide ; utiliser IS NULL, IS NOT NULL et coalesce.

Conversions tolérantes avec TRY_CAST

CAST fait normalement échouer la requête si une valeur ne peut pas être convertie. TRY_CAST retourne plutôt NULL lorsque la combinaison de types est autorisée, mais que la valeur est mal formée ou hors plage :

SELECT TRY_CAST('100$' AS INT) AS amount;  -- NULL
SELECT TRY_CAST('100' AS INT)  AS amount;  -- 100

Équivalent dans une transformation PySpark :

typed_df = raw_df.withColumn(
    "amount",
    F.expr("TRY_CAST(raw_amount AS INT)"),
)

Tester ensuite amount IS NULL permet de mettre les lignes invalides en quarantaine. TRY_CAST ne nettoie pas la chaîne : retirer explicitement un symbole avec regexp_replace seulement si le contrat métier l'autorise.

Timestamps : utiliser TIMESTAMP et normaliser en UTC

Stocker un instant en type TIMESTAMP, pas en chaîne formatée, et normaliser les sources en UTC. Le type reste triable, comparable et utilisable dans les fenêtres, watermarks et agrégations temporelles. Un TIMESTAMP Spark représente un instant ; son affichage dépend du fuseau de la session. Fixer la session en UTC rend les conversions et les résultats reproductibles :

SET TIME ZONE 'UTC';

SELECT
  CAST('2026-08-10T14:30:00Z' AS TIMESTAMP) AS event_ts_utc,
  to_utc_timestamp(
    to_timestamp(local_event_ts, 'yyyy-MM-dd HH:mm:ss'),
    'Europe/Paris'
  ) AS normalized_event_ts
FROM dev.bronze.events;

from pyspark.sql import functions as F

spark.conf.set("spark.sql.session.timeZone", "UTC")

events_typed_df = events_df.withColumn(
    "event_ts",
    F.to_utc_timestamp(
        F.to_timestamp("local_event_ts", "yyyy-MM-dd HH:mm:ss"),
        "Europe/Paris",
    ),
)

Si la chaîne source contient déjà un offset ou Z, la caster directement en TIMESTAMP suffit généralement : ne pas appliquer une seconde conversion de fuseau. Utiliser TIMESTAMP_NTZ seulement pour une heure civile volontairement sans instant universel, par exemple « ouverture du magasin à 09:00 », et non pour un événement distribué.

4.1.3 Données semi-structurées et collections

Décoder les données semi-structurées

Le parcours Bronze → Silver est : identifier le type source → décoder si nécessaire → choisir STRUCT, VARIANT ou une extraction ponctuelle → typer les champs métier.

1. Identifier la source et choisir la stratégie Silver

SELECT typeof(payload) FROM dev.bronze.events LIMIT 1;
SELECT decode(value, 'UTF-8') AS payload_json
FROM dev.bronze.kafka_events;

from pyspark.sql import functions as F

events_df = spark.table("dev.bronze.events")
events_df.printSchema()
kafka_json_df = kafka_df.select(
    F.decode("value", "UTF-8").alias("payload_json")
)

Pour un contrat stable, parser une seule fois vers un STRUCT. Pour une exploration ponctuelle, : est plus concis. Il ne supprime toutefois pas le parsing : Databricks doit interpréter le JSON STRING à chaque requête. Un VARIANT n'apporte un gain de réutilisation que s'il est persisté, pas s'il est créé dans une CTE puis immédiatement supprimé. Le type VARIANT est disponible à partir de Databricks Runtime 15.3.

2A. Schéma stable : schema_of_json, puis from_json

schema_of_json aide à découvrir un schéma DDL sur un exemple ; schema_of_json_agg peut combiner plusieurs documents. En production, relire et versionner un schéma explicite.

-- Exploration uniquement.
SELECT schema_of_json(
  '{"order_id":101,"customer":{"id":7,"address":{"city":"Paris","country":"FR"}},"items":[{"sku":"A1","qty":2},{"sku":"B2","qty":1}]}'
);

-- Parsing de production avec contrat explicite.
CREATE OR REPLACE TABLE dev.silver.order_items AS
WITH parsed AS (
  SELECT from_json(
    payload_json,
    'order_id BIGINT,
     customer STRUCT<
       id: BIGINT,
       address: STRUCT<city: STRING, country: STRING>
     >,
     items ARRAY<STRUCT<sku: STRING, qty: INT>>'
  ) AS order_data
  FROM dev.bronze.orders
)
SELECT
  order_data.order_id,
  order_data.customer.id AS customer_id,
  order_data.customer.address.city AS city,
  item.sku,
  item.qty
FROM parsed
LATERAL VIEW OUTER explode(order_data.items) exploded AS item;

from pyspark.sql import types as T

json_sample = """
{
  "order_id": 101,
  "customer": {
    "id": 7,
    "address": {"city": "Paris", "country": "FR"}
  },
  "items": [
    {"sku": "A1", "qty": 2},
    {"sku": "B2", "qty": 1}
  ]
}
"""

schema_ddl = (
    spark.range(1)
         .select(F.schema_of_json(F.lit(json_sample)).alias("schema"))
         .first()["schema"]
)
print(schema_ddl)

order_schema = T.StructType([
    T.StructField("order_id", T.LongType()),
    T.StructField("customer", T.StructType([
        T.StructField("id", T.LongType()),
        T.StructField("address", T.StructType([
            T.StructField("city", T.StringType()),
            T.StructField("country", T.StringType()),
        ])),
    ])),
    T.StructField("items", T.ArrayType(T.StructType([
        T.StructField("sku", T.StringType()),
        T.StructField("qty", T.IntegerType()),
    ]))),
])

parsed_df = (
    spark.table("dev.bronze.orders")
         .withColumn("order_data", F.from_json("payload_json", order_schema))
         .withColumn("item", F.explode_outer("order_data.items"))
         .select(
             "order_data.order_id",
             F.col("order_data.customer.id").alias("customer_id"),
             F.col("order_data.customer.address.city").alias("city"),
             "item.sku",
             "item.qty",
         )
)

(
    parsed_df.write
             .mode("overwrite")
             .saveAsTable("dev.silver.order_items")
)

Dans le schéma, un dictionnaire JSON devient un STRUCT, une liste devient un ARRAY et chaque objet de la liste devient ici un STRUCT. L'accès utilise . pour les champs imbriqués ; explode_outer transforme chaque élément de items en ligne tout en conservant les commandes dont la liste est NULL ou vide.

Par défaut, from_json utilise un mode permissif et renvoie NULL pour une chaîne non parsable. L'option map('mode', 'FAILFAST') en SQL — ou options={"mode": "FAILFAST"} en PySpark — fait au contraire échouer le traitement. Les noms du schéma doivent respecter la casse des champs JSON. Conserver la donnée brute en Bronze et isoler les erreurs si nécessaire.

2B. Schéma flexible : VARIANT

L'objectif est de convertir le JSON brut une fois pendant le traitement Bronze → Silver, puis de stocker son encodage VARIANT pour les lectures suivantes.

CREATE OR REPLACE TABLE dev.silver.iot_events_variant AS
SELECT
  event_id,
  try_parse_json(payload_json) AS payload
FROM dev.bronze.iot_events;

-- Requête ultérieure : payload est déjà stocké en VARIANT.
SELECT
  payload:device_id::STRING AS device_id,
  TRY_CAST(payload:temperature AS DOUBLE) AS temperature
FROM dev.silver.iot_events_variant;

variant_silver_df = (
    spark.table("dev.bronze.iot_events")
         .withColumn("payload", F.try_parse_json("payload_json"))
         .select("event_id", "payload")
)

(
    variant_silver_df.write
                     .mode("overwrite")
                     .saveAsTable("dev.silver.iot_events_variant")
)

# Lecture ultérieure du VARIANT déjà stocké.
result_df = (
    spark.table("dev.silver.iot_events_variant")
         .select(
             F.expr("payload:device_id::STRING").alias("device_id"),
             F.expr("try_cast(payload:temperature AS DOUBLE)")
              .alias("temperature"),
         )
)

parse_json échoue sur un JSON invalide ; try_parse_json renvoie NULL, ce qui facilite la mise en quarantaine. Sur un VARIANT, : renvoie une valeur VARIANT à caster et les chemins sont sensibles à la casse. Créer ce VARIANT dans une simple CTE ne dispense pas de le reconstruire à la prochaine requête.

2C. Requête ponctuelle dans un JSON STRING

Syntaxe : concise en Databricks SQL :

SELECT
  payload_json:device_id AS device_id,
  TRY_CAST(payload_json:temperature AS DOUBLE) AS temperature
FROM dev.bronze.iot_events;

Alternative avec get_json_object et un chemin JSON $ :

SELECT
  get_json_object(payload_json, '$.device_id') AS device_id,
  TRY_CAST(
    get_json_object(payload_json, '$.temperature') AS DOUBLE
  ) AS temperature
FROM dev.bronze.iot_events;

En PySpark, get_json_object évite d'écrire une expression SQL :

few_fields_df = (
    spark.table("dev.bronze.iot_events")
         .select(
             F.get_json_object("payload_json", "$.device_id")
              .alias("device_id"),
             F.expr("""
                 try_cast(
                   get_json_object(payload_json, '$.temperature')
                   AS DOUBLE
                 )
             """)
              .alias("temperature"),
         )
)

Sur un JSON STRING, les deux méthodes extraient une représentation STRING sans matérialiser de VARIANT. Le parsing est implicite et sera refait lors d'une autre requête. : est généralement plus lisible en Databricks SQL ; get_json_object fournit une fonction explicite en SQL et dans l'API PySpark, et renvoie NULL si le JSON est invalide. Ce choix convient à l'exploration ou à une requête isolée, pas à la matérialisation répétée de nombreux champs Silver.

2D. Modèle hybride recommandé

Pour une Silver flexible, matérialiser les champs fréquemment filtrés, joints ou agrégés en colonnes typées et conserver le document complet en VARIANT pour les champs rares ou évolutifs :

CREATE OR REPLACE TABLE dev.silver.iot_events AS
SELECT
  payload_json:device_id AS device_id,
  TRY_CAST(payload_json:temperature AS DOUBLE) AS temperature,
  try_parse_json(payload_json) AS payload
FROM dev.bronze.iot_events;

hybrid_silver_df = (
    spark.table("dev.bronze.iot_events")
         .select(
             F.expr("payload_json:device_id").alias("device_id"),
             F.expr("try_cast(payload_json:temperature AS DOUBLE)")
              .alias("temperature"),
             F.try_parse_json("payload_json").alias("payload"),
         )
)

(
    hybrid_silver_df.write
                    .mode("overwrite")
                    .saveAsTable("dev.silver.iot_events")
)

Une colonne VARIANT ne peut pas être utilisée directement pour partitionner, ZORDER, définir une clé de clustering, comparer, grouper ou trier. Extraire les champs fréquents en colonnes typées permet ces opérations et améliore le layout ; le VARIANT conserve la flexibilité du reste du document.

En pratique : 2A pour un schéma stable ; 2B pour un document flexible persisté ; 2C pour une requête ponctuelle ; 2D pour combiner performances sur les champs fréquents et flexibilité. Conserver aussi le STRING brut en Bronze pour l'audit et le rejeu.

Manipuler les colonnes et les tableaux

Version SQL :

SELECT DISTINCT
  id,
  split(full_name, ' ') AS name_parts,
  tag
FROM dev.bronze.people
LATERAL VIEW OUTER explode(tags) exploded AS tag
WHERE tag <> 'obsolete';

Version PySpark :

result_df = (
    source_df
      .select("id", "full_name", "tags")
      .withColumn("name_parts", F.split("full_name", " "))
      .withColumn("tag", F.explode_outer("tags"))
      .filter(F.col("tag") != "obsolete")
      .drop("tags")
      .dropDuplicates(["id", "tag"])
)

explode() ou flatten()

• explode(array) transforme chaque élément d'un tableau en une nouvelle ligne : le nombre de lignes augmente. explode_outer conserve une ligne avec NULL pour un tableau nul ou vide.
• posexplode(array) fait la même expansion et retourne aussi la position d'origine de chaque élément (pos, indexé à partir de zéro).
• flatten(array_of_arrays) fusionne un niveau de tableaux imbriqués en un seul tableau : le nombre de lignes ne change pas.

SELECT order_id, explode(items) AS item
FROM dev.bronze.orders;

SELECT order_id, pos, item
FROM dev.bronze.orders,
LATERAL posexplode(items) AS p(pos, item);

SELECT order_id, flatten(item_batches) AS all_items
FROM dev.bronze.orders;

exploded_df = orders_df.select("order_id", F.explode("items").alias("item"))
positioned_df = orders_df.select(
    "order_id",
    F.posexplode("items").alias("pos", "item"),
)
flattened_df = orders_df.select(
    "order_id", F.flatten("item_batches").alias("all_items")
)

Mémo : explode = plus de lignes ; posexplode = lignes + positions ; flatten = tableau moins imbriqué.

Filtrer les éléments d'un tableau avec une lambda

La fonction higher-order FILTER conserve la ligne et retourne un nouveau tableau contenant uniquement les éléments qui satisfont le prédicat. Elle ne doit pas être confondue avec la clause WHERE, qui filtre des lignes, ni avec explode, qui crée des lignes :

SELECT
  faculty_id,
  students,
  FILTER(students, student -> student.total_courses < 3) AS few_courses_students
FROM dev.silver.faculties;

from pyspark.sql import functions as F

faculties_df.select(
    "faculty_id",
    "students",
    F.filter(
        "students",
        lambda student: student.total_courses < 3,
    ).alias("few_courses_students"),
)

Le résultat reste un ARRAY<STRUCT<...>>; pour obtenir une ligne par étudiant, appliquer ensuite explode ou explode_outer.

4.1.4 Déduplication déterministe

Pour supprimer des lignes entièrement identiques, appliquer DISTINCT à toutes les colonnes :

SELECT DISTINCT *
FROM dev.bronze.orders;

unique_rows_df = source_df.distinct()

distinct() compare toutes les colonnes et peut provoquer un shuffle. dropDuplicates(["id"]) déduplique seulement selon la clé indiquée, mais ne permet pas de choisir explicitement la ligne conservée. Pour garder, par exemple, la version la plus récente de chaque clé, utiliser une fenêtre déterministe :

Version SQL :

SELECT * EXCEPT (rn)
FROM (
  SELECT *, row_number() OVER (
    PARTITION BY order_id
    ORDER BY updated_at DESC, ingested_at DESC
  ) AS rn
  FROM dev.bronze.orders
)
WHERE rn = 1;

Version PySpark :

w = Window.partitionBy("order_id").orderBy(
    F.col("updated_at").desc(),
    F.col("ingested_at").desc()
)

latest_df = (
    source_df.withColumn("rn", F.row_number().over(w))
             .filter(F.col("rn") == 1)
             .drop("rn")
)

4.1.5 Écriture et évolution des tables Delta (DDL et DML)

DDL — Data Definition Language (langage de définition des données) définit les objets et leur structure : CREATE, ALTER, DROP, TRUNCATE, CREATE VIEW.

DML — Data Manipulation Language (langage de manipulation des données) lit ou modifie les lignes : SELECT, INSERT, UPDATE, DELETE, MERGE. Selon les classifications, SELECT est parfois rangé à part dans le DQL ; pour l'examen, retenir surtout que le DML agit sur les données, le DDL sur la structure.

-- Remplace la définition et crée une table vide.
CREATE OR REPLACE TABLE dev.silver.employees (
  employee_id STRING,
  start_date DATE,
  avg_rating DOUBLE
) USING DELTA;

INSERT INTO dev.silver.employees VALUES ('a1', DATE '2026-01-06', 9.4);

UPDATE dev.silver.employees
SET avg_rating = 9.7
WHERE employee_id = 'a1';

DELETE FROM dev.silver.employees
WHERE employee_id = 'a1';

Différence importante :

• CREATE TABLE IF NOT EXISTS ne modifie rien si la table existe ;
• CREATE OR REPLACE TABLE remplace la définition et les données de la table ;
• CTAS crée/remplace une table avec le résultat d'un SELECT.

Le mode d'écriture PySpark ignore est proche de IF NOT EXISTS : si la table ou le chemin existe déjà, l'écriture entière est silencieusement ignorée.

CREATE TABLE IF NOT EXISTS dev.silver.orders_snapshot
USING DELTA
AS SELECT * FROM dev.silver.orders;

(orders_df.write
          .format("delta")
          .mode("ignore")
          .saveAsTable("dev.silver.orders_snapshot"))

Ce mode ne déduplique pas les lignes et ne vérifie pas que la table existante est complète ou conforme : il convient à la création idempotente d'un objet, pas à un chargement incrémental. Pour un upsert par clé, utiliser MERGE.

Le DDL reste principalement SQL. Pour les DML programmatiques, l'API DeltaTable expose update, delete et merge ; la section suivante montre le cas merge, le plus important pour l'examen.

Colonnes générées

Une colonne GENERATED ALWAYS AS (expression) est calculée par Delta à l'écriture, puis stockée physiquement, à partir d'autres colonnes de la même ligne. Elle évite de fournir manuellement une dérivation déterministe, par exemple la date issue d'un timestamp :

CREATE TABLE dev.silver.orders (
  order_id INT,
  order_time TIMESTAMP,
  order_date DATE GENERATED ALWAYS AS (CAST(order_time AS DATE)),
  units INT
) USING DELTA;

-- order_date est calculée automatiquement à partir de order_time.
INSERT INTO dev.silver.orders (order_id, order_time, units)
VALUES (1, TIMESTAMP '2026-08-18 09:30:00', 4);

Version Python avec le builder Delta :

from delta.tables import DeltaTable
from pyspark.sql.types import DateType

(DeltaTable.create(spark)
 .tableName("dev.silver.orders")
 .addColumn("order_id", "INT")
 .addColumn("order_time", "TIMESTAMP")
 .addColumn(
     "order_date",
     DateType(),
     generatedAlwaysAs="CAST(order_time AS DATE)",
 )
 .addColumn("units", "INT")
 .execute())

À retenir : le type de la colonne générée est déclaré explicitement ; l'expression doit être déterministe et ne peut pas être une agrégation, une sous-requête ou une UDF. Une écriture peut omettre order_date ; si elle fournit cette colonne, sa valeur doit respecter l'expression, sinon la transaction échoue. La colonne peut servir au partitionnement, mais pour une nouvelle table managée il faut également considérer liquid clustering et éviter un partitionnement excessif.

Évolution du schéma avec mergeSchema et overwriteSchema

Ces options concernent le schéma de la table Delta écrite, et non le schéma d'entrée conservé dans cloudFiles.schemaLocation. Elles s'appliquent indépendamment du fait que le DataFrame ait été produit en batch ou par une ingestion streaming.

Version SQL, avec une évolution explicite ou un remplacement contrôlé :

-- Évolution additive explicite, sans réécrire les lignes existantes.
ALTER TABLE dev.silver.orders ADD COLUMNS (sales_channel STRING);

-- Remplacement complet : données et schéma deviennent ceux du SELECT.
CREATE OR REPLACE TABLE dev.silver.orders AS
SELECT CAST(order_id AS BIGINT) AS order_id,
       CAST(amount AS DECIMAL(14,2)) AS amount
FROM dev.silver.orders_retyped;

Version PySpark avec les options d'écriture DataFrame :

# Ajout de colonnes compatibles sans remplacer les lignes déjà présentes.
(new_orders_df.write
              .format("delta")
              .mode("append")
              .option("mergeSchema", "true")
              .saveAsTable("dev.silver.orders"))

# Remplacement volontaire des données ET du schéma de la table.
(retyped_orders_df.write
                  .format("delta")
                  .mode("overwrite")
                  .option("overwriteSchema", "true")
                  .saveAsTable("dev.silver.orders"))

À retenir : mergeSchema élargit de façon additive ; overwriteSchema accompagne un overwrite destructif. Ni l'un ni l'autre ne remplace un contrat de données relu. Pour un MERGE INTO, utiliser plutôt WITH SCHEMA EVOLUTION ou .withSchemaEvolution() dans la section suivante.

La configuration de session suivante active l'évolution de schéma pour toutes les écritures Delta compatibles de la SparkSession :

-- Consulter la valeur de la session.
SET spark.databricks.delta.schema.autoMerge.enabled;

-- L'activer pour les écritures Delta compatibles de cette session.
SET spark.databricks.delta.schema.autoMerge.enabled = true;

# Consulter, puis modifier la valeur pour la SparkSession courante.
spark.conf.get("spark.databricks.delta.schema.autoMerge.enabled", "false")
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")

spark.databricks.delta.schema.autoMerge.enabled est un réglage global à la session, pas une propriété d'une seule table ni une option Auto Loader. Il peut donc faire évoluer involontairement plusieurs cibles. Databricks le considère comme une approche legacy et recommande en production d'autoriser l'évolution au niveau de l'opération concernée : .option("mergeSchema", "true"), INSERT WITH SCHEMA EVOLUTION, MERGE WITH SCHEMA EVOLUTION ou .withSchemaEvolution(). L'option ou la syntaxe locale est prioritaire sur la configuration de session.

Pour imposer une table insert-only, activer la propriété Delta appendOnly :

ALTER TABLE dev.bronze.events SET TBLPROPERTIES (
  'delta.appendOnly' = 'true'
);

# Cette propriété de table reste un DDL SQL, même depuis un notebook Python.
spark.sql("""
  ALTER TABLE dev.bronze.events SET TBLPROPERTIES (
    'delta.appendOnly' = 'true'
  )
""")

Les INSERT et appends restent permis ; les opérations qui modifient ou suppriment des lignes existantes (UPDATE, DELETE, MERGE concerné) échouent. Cette propriété protège un journal immuable, mais ne déduplique pas les inserts.

Nuance de QCM : ce n'est pas le mot-clé MERGE qui est interdit. Un insert-only merge, ne comportant que WHEN NOT MATCHED THEN INSERT, reste compatible avec appendOnly puisqu'il ne modifie aucune ligne existante. Un MERGE comportant un UPDATE ou un DELETE échoue.

Contraintes NOT NULL et CHECK Delta Lake

Pour rendre non nullable une colonne existante :

-- À contrôler/corriger avant la modification.
SELECT COUNT(*) AS null_ids
FROM dev.silver.transactions
WHERE transaction_id IS NULL;

ALTER TABLE dev.silver.transactions
ALTER COLUMN transaction_id SET NOT NULL;

SET NOT NULL est pris en charge pour une table Delta. Databricks vérifie toutes les lignes existantes avant d'ajouter la contrainte : si au moins une valeur est NULL, l'ALTER TABLE échoue et il faut corriger, supprimer ou mettre en quarantaine ces lignes. Les écritures futures contenant un NULL font ensuite échouer toute leur transaction. DROP NOT NULL rend à nouveau la colonne nullable :

ALTER TABLE dev.silver.transactions
ALTER COLUMN transaction_id DROP NOT NULL;

Une contrainte CHECK est une règle appliquée à chaque écriture. Databricks vérifie aussi les lignes existantes lorsqu'on ajoute la contrainte. Si une ligne viole la règle, toute la transaction échoue : aucune écriture partielle n'est validée.

ALTER TABLE dev.silver.orders
ADD CONSTRAINT valid_amount CHECK (amount >= 0);

-- Échec atomique si une ligne a amount < 0.
INSERT INTO dev.silver.orders VALUES (1001, DATE '2026-08-07', -5);

ALTER TABLE dev.silver.orders DROP CONSTRAINT valid_amount;

En PySpark, la contrainte reste définie en DDL SQL, puis elle s'applique automatiquement aux écritures DataFrame :

spark.sql("""
  ALTER TABLE dev.silver.orders
  ADD CONSTRAINT valid_amount CHECK (amount >= 0)
""")

# Cette écriture échoue entièrement si invalid_df contient amount < 0.
invalid_df.write.mode("append").saveAsTable("dev.silver.orders")

Les définitions des contraintes CHECK et NOT NULL font partie des métadonnées transactionnelles enregistrées dans le Delta log ; chaque commit les applique atomiquement. Elles sont donc enforced. Les clés primaires et étrangères Unity Catalog sont informatives : ne pas leur attribuer le même comportement de rejet.

4.1.6 Upsert avec MERGE

MERGE INTO dev.silver.customers AS target
USING dev.bronze.customer_changes AS source
ON target.customer_id = source.customer_id

WHEN MATCHED AND source.operation = 'DELETE' THEN DELETE

WHEN MATCHED AND source.operation = 'UPDATE' THEN
  UPDATE SET
    target.email = source.email,
    target.country = source.country,
    target.updated_at = source.updated_at

WHEN NOT MATCHED AND source.operation <> 'DELETE' THEN
  INSERT (customer_id, email, country, updated_at)
  VALUES (source.customer_id, source.email, source.country, source.updated_at);

Version PySpark avec l'API Delta :

from delta.tables import DeltaTable

target = DeltaTable.forName(spark, "dev.silver.customers")
source_df = spark.table("dev.bronze.customer_changes")

(target.alias("target")
 .merge(source_df.alias("source"),
        "target.customer_id = source.customer_id")
 .whenMatchedDelete(condition="source.operation = 'DELETE'")
 .whenMatchedUpdate(
     condition="source.operation = 'UPDATE'",
     set={
         "email": "source.email",
         "country": "source.country",
         "updated_at": "source.updated_at",
     },
 )
 .whenNotMatchedInsert(
     condition="source.operation <> 'DELETE'",
     values={
         "customer_id": "source.customer_id",
         "email": "source.email",
         "country": "source.country",
         "updated_at": "source.updated_at",
     },
 )
 .execute())

Avec évolution de schéma compatible :

MERGE WITH SCHEMA EVOLUTION INTO dev.silver.customers AS target
USING dev.bronze.customer_changes AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Équivalent PySpark à partir de Databricks Runtime 15.4 LTS :

(target.alias("target")
 .merge(source_df.alias("source"),
        "target.customer_id = source.customer_id")
 .withSchemaEvolution()
 .whenMatchedUpdateAll()
 .whenNotMatchedInsertAll()
 .execute())

SQL rend généralement un MERGE plus facile à relire. L'API Delta PySpark est pratique lorsque la source est déjà un DataFrame ou que les clauses sont construites dans du code modulaire.

À retenir :

• MERGE est la primitive d'upsert et de CDC sur une table Delta ;
• WITH SCHEMA EVOLUTION autorise le MERGE à faire évoluer le schéma cible pour les colonnes source compatibles, notamment avec UPDATE SET * et INSERT * ;
• sans cette clause, une colonne source absente de la cible provoque normalement une erreur ou doit être ignorée explicitement ;
• l'évolution du schéma du MERGE est distincte de celle d'Auto Loader : la première concerne la table cible pendant l'upsert, la seconde le schéma découvert à la lecture de fichiers ;
• la condition ON exprime la clé métier ;
• plusieurs lignes source correspondant à une même ligne cible peuvent rendre le merge ambigu : dédupliquer la source d'abord ;
• une clé de MERGE doit être stable et unique au niveau métier : plusieurs lignes source pour une même clé peuvent provoquer une erreur si elles modifient la même cible, et plusieurs lignes source nouvelles portant la même clé peuvent toutes être insérées ;
• l'évolution de schéma n'est pas une validation métier : contrôler les colonnes autorisées.

Pour un MERGE qui lit trop de fichiers cible, commencer par vérifier la condition ON et le plan. Ajouter un prédicat borné sur une partition ou une date seulement s'il est sémantiquement correct, puis améliorer le layout sur la clé souvent recherchée :

-- Le filtre de date est valide ici parce que le micro-batch ne contient que ce jour.
MERGE INTO dev.silver.events AS target
USING dev.staging.events_day AS source
ON  target.event_id = source.event_id
AND target.event_date = source.event_date
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

-- Table non liquid-clustered : rapproche les mêmes clés et améliore le data skipping.
OPTIMIZE dev.silver.events ZORDER BY (event_id);

from delta.tables import DeltaTable

(DeltaTable.forName(spark, "dev.silver.events")
 .alias("target")
 .merge(
     source_df.alias("source"),
     "target.event_id = source.event_id "
     "AND target.event_date = source.event_date",
 )
 .whenMatchedUpdateAll()
 .whenNotMatchedInsertAll()
 .execute())

Le Z-order n'est pas une garantie de pruning et ne corrige pas une mauvaise clé : il rend les statistiques min/max plus sélectives afin que le data skipping automatique ignore davantage de fichiers. Pour une nouvelle table managée, considérer d'abord liquid clustering. Les petits fichiers produits par des MERGE fréquents relèvent plutôt des optimized writes/auto compaction, puis d'OPTIMIZE pour l'existant ; voir 7.5.3.

4.1.7 Data quality expectations

Les expectations sont des règles de qualité propres aux Lakeflow Spark Declarative Pipelines, présentés plus largement en 4.3. Elles sont souvent appliquées pendant Bronze → Silver pour mesurer, écarter ou bloquer les données invalides.

CREATE OR REFRESH STREAMING TABLE dev.silver.orders (
  CONSTRAINT valid_id
    EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW,
  CONSTRAINT valid_amount
    EXPECT (amount >= 0),
  CONSTRAINT recent_date
    EXPECT (order_ts >= TIMESTAMP '2020-01-01') ON VIOLATION FAIL UPDATE
)
COMMENT 'Commandes nettoyées'
TBLPROPERTIES ('quality' = 'silver')
AS
SELECT
  CAST(order_id AS BIGINT) AS order_id,
  CAST(order_ts AS TIMESTAMP) AS order_ts,
  CAST(amount AS DECIMAL(12,2)) AS amount
FROM STREAM(dev.bronze.orders);

Version Python avec les trois comportements :