Ingérer des données avec le chargeur automatique

Lorsque de nouveaux fichiers de données arrivent en continu dans le stockage cloud, vous avez besoin d’un moyen efficace de les traiter sans suivre manuellement les fichiers qui ont été ingérés. Auto Loader résout ce problème en détectant et en ingérant automatiquement les nouveaux fichiers dès leur apparition, en offrant des garanties d'exactement une fois, sans vous obliger à gérer manuellement l'état ou les checkpoints.

Comprendre le fonctionnement du chargeur automatique

Le chargeur automatique est une source de diffusion en continu structurée qui surveille un emplacement de stockage cloud et traite de manière incrémentielle les nouveaux fichiers. Au lieu d’analyser des répertoires entiers à plusieurs reprises, le chargeur automatique conserve l’état des fichiers qu’il a déjà traités, ce qui garantit que chaque fichier est ingéré exactement une seule fois.

Lorsque vous démarrez un flux de chargeur automatique, il peut traiter les fichiers existants dans le répertoire, puis surveiller en permanence les nouveaux arrivées. Le flux stocke les informations de progression dans un emplacement de point de contrôle, ce qui lui permet de reprendre exactement à partir de l’endroit où il s’est arrêté s’il est interrompu.

Le chargeur automatique détecte les nouveaux fichiers à l’aide de l’un des deux modes suivants :

• Mode de liste de répertoires : le chargeur automatique répertorie régulièrement le répertoire d’entrée pour découvrir de nouveaux fichiers. Cette approche ne nécessite aucune configuration supplémentaire au-delà de l’accès au stockage.
• Mode de notification de fichier : le chargeur automatique utilise les services de notification cloud pour recevoir des événements lorsque de nouveaux fichiers arrivent. Ce mode est plus efficace pour les charges de travail à grande échelle, car il évite les listes répétées d’annuaires.

Pour la plupart des charges de travail de production, Databricks recommande le mode de notification de fichier avec les événements de fichier activés sur votre emplacement externe dans le catalogue Unity. Avec les événements de fichier, le chargeur automatique reçoit des notifications directement lorsque les fichiers arrivent, ce qui réduit la latence et les coûts d’API cloud.

Configurer le chargeur automatique pour l’ingestion de fichiers

Pour ingérer des données avec le chargeur automatique, vous utilisez le cloudFiles format avec spark.readStream. L’exemple suivant lit les fichiers JSON à partir d’Azure Data Lake Storage et les écrit dans une table de catalogue Unity :

base_path = "abfss://container@storage.dfs.core.windows.net/autoloader/orders"
schema_path = f"{base_path}/schema"
checkpoint_path = f"{base_path}/checkpoint"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", schema_path)
  .load("abfss://container@storage.dfs.core.windows.net/incoming/orders/")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("sales.bronze.orders"))

L’option cloudFiles.format spécifie le format de fichier source. Le chargeur automatique prend en charge les fichiers JSON, CSV, Parquet, Avro, ORC, XML, texte et binaire. Le cloudFiles.schemaLocation stocke le schéma déduit, ce qui permet au Chargeur Automatique de suivre les modifications de schéma au fil du temps. Bien que l’emplacement du schéma et l’emplacement du point de contrôle puissent être identiques, leur séparation vous permet de réinitialiser un flux sans perdre le schéma déduit.

Le trigger(availableNow=True) paramètre traite tous les fichiers actuellement disponibles, puis s’arrête. Ce modèle fonctionne bien pour les travaux de traitement par lots planifiés. Pour le traitement continu, vous pouvez omettre le déclencheur ou l’utiliser trigger(processingTime="1 minute") pour traiter des fichiers à intervalles.

Utiliser la syntaxe SQL avec read_files

Auto Loader prend également en charge la syntaxe SQL via la fonction tabulaire read_files. Cette approche est utile lorsque vous préférez SQL ou lorsque vous travaillez dans des notebooks SQL :

SELECT * FROM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json',
  schemaHints => 'order_id INT, amount DECIMAL(10,2)'
)

Lorsque vous utilisez read_files avec des tables de streaming dans des pipelines Lakeflow, les fonctionnalités d’Auto Loader sont automatiquement activées. Utilisez le mot clé pour activer le STREAM traitement incrémentiel :

CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json'
)

Dans ce contexte, le pipeline gère automatiquement le checkpointing et l'évolution du schéma.

Gérer l’inférence et l’évolution du schéma

Le chargeur automatique peut détecter automatiquement le schéma de vos fichiers sources, ce qui élimine la nécessité de définir manuellement des schémas. Lorsque l’inférence de schéma est activée, le chargeur automatique échantillonne des fichiers pour déterminer les noms de colonnes et les types de données.

Pour les fichiers JSON et CSV, le chargeur automatique déduit toutes les colonnes sous forme de chaînes par défaut. Cette approche conservatrice empêche les incompatibilités de type susceptibles d’entraîner une perte de données. Pour activer la détection automatique des types, définissez l’option cloudFiles.inferColumnTypes :

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(source_path))

Lorsque vos données sources introduisent de nouvelles colonnes, le chargeur automatique détecte la modification et peut évoluer automatiquement le schéma cible. L’option cloudFiles.schemaEvolutionMode contrôle ce comportement :

La colonne de données sauvée (_rescued_data) capture toutes les données qui ne correspondent pas au schéma attendu, y compris les incompatibilités de type et les colonnes inattendues. Cette fonctionnalité empêche la perte de données lorsque les données sources ne sont pas conformes aux attentes. Le chargeur automatique ajoute automatiquement cette colonne lors de l’inférence du schéma.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .schema(expected_schema)
  .load(source_path))

Pour renommer la colonne de données sauvée, utilisez l’option rescuedDataColumn suivante :

.option("rescuedDataColumn", "my_rescued_data")

Si vous savez que certaines colonnes doivent avoir des types spécifiques, utilisez des indicateurs de schéma pour remplacer les types déduits :

.option("cloudFiles.schemaHints", "order_date DATE, amount DECIMAL(10,2)")

Surveiller et optimiser les flux du chargeur automatique

Pour les charges de travail de production, la surveillance vous aide à suivre la progression de l’ingestion et à identifier les problèmes. Le chargeur automatique fournit une fonction SQL pour interroger l’état des fichiers découverts :

SELECT * FROM cloud_files_state('/path/to/checkpoint');

Cette requête retourne des métadonnées sur les fichiers détectés par le chargeur automatique, y compris leur état de traitement. Vous pouvez utiliser ces informations pour vérifier que les fichiers sont ingérés comme prévu.

Le Auto Loader signale des métriques par le biais du tableau de bord de progression Structured Streaming, notamment :

• numFilesOutstanding: fichiers découverts mais pas encore traités
• numBytesOutstanding: nombre total d’octets en attente de traitement

Pour les charges de travail à volume élevé, contrôlez le taux de traitement à l’aide cloudFiles.maxFilesPerTrigger ou cloudFiles.maxBytesPerTrigger. Ces paramètres empêchent tout micro-lot unique de devenir trop volumineux :

.option("cloudFiles.maxFilesPerTrigger", "1000")
.option("cloudFiles.maxBytesPerTrigger", "1g")

Pour accéder aux métadonnées au niveau du fichier dans vos données ingérées, sélectionnez la _metadata colonne. Cette colonne masquée contient des informations sur chaque fichier source, y compris file_path, file_name, file_sizeet file_modification_time:

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .load(source_path)
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Auto Loader s’intègre en toute transparence aux pipelines Lakeflow, où il gère automatiquement la gestion des points de contrôle et des schémas. Dans l’unité suivante, vous allez apprendre à utiliser le chargeur automatique dans les définitions de pipeline déclaratives pour les flux de travail d’ingestion de données de bout en bout.