Ingérer des données avec des notebooks

Les notebooks dans Azure Databricks fournissent une approche flexible, pilotée par le code, pour ingérer des données provenant de diverses sources. Lorsque les outils graphiques ne répondent pas à vos besoins ou que vous avez besoin d’une logique personnalisée pour la transformation des données, les notebooks vous permettent de contrôler complètement le processus d’ingestion. Vous pouvez utiliser Python (PySpark), SQL, Scala ou R pour vous connecter à des sources de données, appliquer des transformations et charger des données dans des tables du catalogue Unity.

Dans cette unité, vous allez apprendre à ingérer des données à l’aide de notebooks pour les scénarios de traitement par lots et de streaming.

Ingérer des données par lots avec des DataFrames

Les DataFrames constituent la base de l’ingestion de données par lots dans Databricks. Un DataFrame est une collection distribuée de données organisées en colonnes nommées, semblable à une table d’une base de données relationnelle. Vous utilisez l’API spark.read pour charger des données de différentes sources dans un DataFrame.

L’exemple suivant lit un fichier CSV dans un DataFrame :

df = (spark.read
    .format("csv")
    .option("header", "true")
    .option("inferSchema", "true")
    .load("/Volumes/catalog/schema/volume/data.csv"))

La format() méthode spécifie le type de fichier, tandis que les option() méthodes configurent la façon dont les données sont analysées. L’option header indique à Spark d’utiliser la première ligne comme noms de colonnes et inferSchema détecte automatiquement les types de données.

Lecture à partir de différents formats de fichier

Azure Databricks prend en charge plusieurs formats de fichiers. Vous pouvez ajuster le format et les options en fonction de votre source de données :

# Read JSON files
df_json = spark.read.format("json").load("/path/to/data.json")

# Read Parquet files (columnar format, efficient for analytics)
df_parquet = spark.read.format("parquet").load("/path/to/data.parquet")

# Read XML files (requires rowTag option)
df_xml = (spark.read
    .format("xml")
    .option("rowTag", "record")
    .load("/path/to/data.xml"))

<data>
  <record><id>1</id><name>Alice</name></record>
  <record><id>2</id><name>Bob</name></record>
</data>

Les fichiers Parquet stockent des données dans un format columnar, ce qui améliore les performances des requêtes pour les charges de travail analytiques. Les fichiers JSON prennent en charge les structures imbriquées et sont courants pour les réponses d’API. Les fichiers XML nécessitent l’option rowTag pour spécifier quel élément correspond à chaque ligne.

Se connecter à des bases de données externes

Pour les données stockées dans des bases de données relationnelles, utilisez des connexions JDBC pour lire directement à partir de la source :

df_jdbc = (spark.read
    .format("jdbc")
    .option("url", "jdbc:sqlserver://server.database.windows.net:1433;database=mydb")
    .option("dbtable", "schema.tablename")
    .option("user", dbutils.secrets.get(scope="jdbc", key="username"))
    .option("password", dbutils.secrets.get(scope="jdbc", key="password"))
    .load())

Stockez les informations d'identification dans les secrets Databricks plutôt que de les coder en dur dans les notebooks. Vous pouvez également envoyer des requêtes vers la base de données à l’aide d’une sous-requête dans l’option dbtable :

pushdown_query = "(SELECT id, name, amount FROM orders WHERE order_date > '2024-01-01') AS filtered_orders"
df = spark.read.format("jdbc").option("dbtable", pushdown_query).load()

Écrire des données dans des tables de catalogue Unity

Après avoir chargé et transformé vos données, écrivez-les dans une table de catalogue Unity. Azure Databricks utilise le format Delta Lake par défaut, qui offre des transactions et un versionnage ACID :

df.write.mode("overwrite").saveAsTable("catalog.schema.table_name")

L’option mode() contrôle la façon dont les données existantes sont gérées :

• overwrite : remplace entièrement le contenu de la table
• append : ajoute des lignes à la table existante
• ignorer : ne fait rien si la table existe déjà
• erreur : lève une exception si la table existe (par défaut)

Ingérer des données de streaming avec Structured Streaming

Structured Streaming vous permet de traiter les données en continu à mesure qu’elles arrivent. Au lieu d'utiliser spark.read, vous utilisez spark.readStream pour créer un DataFrame de streaming qui traite les nouvelles données de manière incrémentielle.

L’exemple suivant lit les données de streaming à partir de Kafka :

df_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "topic-name")
    .option("startingOffsets", "latest")
    .load())

L’option startingOffsets détermine où commencer la lecture lors du démarrage d’une nouvelle requête. Utilisez latest pour uniquement les nouvelles données ou earliest pour lire à partir du début du sujet. Une fois que vous avez configuré un emplacement de point de contrôle dans le flux d’écriture, les exécutions suivantes reprendnt automatiquement à partir du dernier décalage traité, en ignorant startingOffsets.

Traiter des flux basés sur des fichiers

Vous pouvez également diffuser en continu à partir de sources de fichiers à mesure que de nouveaux fichiers arrivent dans un répertoire :

df_file_stream = (spark.readStream
    .format("csv")
    .option("header", "true")
    .schema(defined_schema)
    .load("/Volumes/catalog/schema/volume/incoming/"))

Pour les sources de fichiers de streaming, vous devez définir le schéma explicitement plutôt que d’utiliser inferSchema. L’inférence de schéma n’est pas prise en charge pour les sources de streaming, car elle nécessite la lecture de l’ensemble du jeu de données.

Écrire des données de diffusion en continu dans des tables

Utilisez writeStream pour conserver les données de streaming dans les tables du Unity Catalog :

(df_stream
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("catalog.schema.streaming_table"))

Le mode de sortie détermine quels enregistrements sont écrits :

• append : seules les nouvelles lignes sont écrites (par défaut pour la plupart des opérations)
• mise à jour : les lignes modifiées sont enregistrées
• complet : toutes les lignes sont réécrites (utilisées avec des agrégations)

L’emplacement du point de contrôle stocke les informations de progression, ce qui permet au flux de reprendre à partir de l’endroit où il s’est arrêté en cas d’interruption.

Gérer les données semi-structurées

Lorsque vous utilisez des données semi-structurées comme JSON, le type de données VARIANT offre une flexibilité pour stocker et interroger des structures imbriquées :

from pyspark.sql.functions import parse_json, variant_get, col

df = spark.read.text("/path/to/data.json")
df_variant = df.select(parse_json(col("value")).alias("data"))

Vous pouvez ensuite interroger des champs spécifiques à l’aide de la fonction ou de la variant_get syntaxe SQL :

df_variant.select(variant_get(col("data"), "$.customer.name", "string")).display()

Cette approche fonctionne bien lorsque le schéma JSON varie entre les enregistrements ou évolue au fil du temps.