Ingérer des données avec Spark Structured Streaming

L’ingestion de données en temps réel vous permet de traiter les événements au fur et à mesure qu’ils se produisent plutôt que d’attendre les intervalles de traitement par lots. Lorsque votre organisation doit capturer des relevés de capteurs IoT, des données de clickstream ou des transactions financières au fur et à mesure qu'elles se produisent, Spark Structured Streaming fournit la base d'une ingestion continue des données dans votre lakehouse.

Dans cette unité, vous allez apprendre à configurer des travaux Structured Streaming qui lisent à partir de sources de streaming et écrivent dans des tables de catalogue Unity.

Comprendre le modèle Structured Streaming

Structured Streaming traite les données entrantes sous la forme d’une table non entrante qui augmente en continu à mesure que de nouvelles données arrivent. Vous utilisez la même API DataFrame que vous connaissez déjà, mais au lieu de spark.readcela, vous utilisez spark.readStream pour créer un DataFrame de streaming qui traite les données de manière incrémentielle.

Envisagez un scénario dans lequel les données de capteur arrivent en continu à partir de l’équipement de fabrication. Au lieu de planifier des travaux par lots horaires pour ingérer ces données, vous pouvez configurer un travail de streaming qui traite chaque lecture en quelques secondes de son arrivée. Le moteur de diffusion en continu suit les enregistrements qui ont été traités à l’aide de points de contrôle. Par conséquent, si le travail redémarre, il reprend à partir de l’endroit où il s’est arrêté sans retraiter ou manquer de données.

Le modèle fondamental pour l’ingestion de streaming implique trois étapes :

1. Configurer la source à l'aide de spark.readStream avec les détails de connexion
2. Appliquer des transformations pour mettre en forme les données selon les besoins
3. Écrire vers un récepteur à l'aide de writeStream avec un emplacement de checkpoint

Se connecter à Kafka et Event Hubs

Azure Event Hubs fournit un point de terminaison compatible Apache Kafka, ce qui signifie que vous pouvez utiliser le même connecteur Kafka pour les deux services. Cette approche unifiée simplifie votre code lors de l’utilisation de différents bus de messages.

L'exemple suivant lit des messages à partir d'une rubrique Kafka :

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

Pour Azure Event Hubs, vous vous connectez via son interface compatible Kafka. La principale différence est la configuration d’authentification, qui utilise SASL avec votre chaîne de connexion Event Hubs :

eh_namespace = "your-namespace"
eh_name = "your-eventhub"
connection_string = dbutils.secrets.get(scope="eventhubs", key="connection-string")

kafka_options = {
    "kafka.bootstrap.servers": f"{eh_namespace}.servicebus.windows.net:9093",
    "subscribe": eh_name,
    "kafka.sasl.mechanism": "PLAIN",
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.jaas.config": f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{connection_string}";'
}

df_stream = (spark.readStream
    .format("kafka")
    .options(**kafka_options)
    .load())

Lors de la lecture à partir de Kafka ou d’Event Hubs via l’interface compatible Kafka, vous recevez un schéma standard avec key, , value, topicpartition, offsetet timestamp des colonnes. Les colonnes key et value contiennent des données binaires, vous les transformez donc généralement en chaîne et en parsez le contenu :

from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StringType, DoubleType

schema = StructType() \
    .add("device_id", StringType()) \
    .add("temperature", DoubleType()) \
    .add("timestamp", StringType())

parsed_df = (df_stream
    .select(col("value").cast("string").alias("json_value"))
    .select(from_json(col("json_value"), schema).alias("data"))
    .select("data.*"))

Utiliser le connecteur Event Hubs natif

Azure Databricks fournit également un connecteur Event Hubs natif qui simplifie l’authentification et fournit un accès direct aux fonctionnalités spécifiques à Event Hubs. Ce connecteur utilise le eventhubs format suivant :

connection_string = dbutils.secrets.get(scope="eventhubs", key="connection-string")

events = (spark.readStream
    .format("eventhubs")
    .option("eventhubs.connectionString", connection_string)
    .load())

Le connecteur Event Hubs natif retourne un schéma différent de l’interface compatible Kafka. La charge utile d’événement est stockée dans la body colonne (au lieu de value), ainsi que d’autres colonnes de métadonnées spécifiques à Event Hubs :

# Extract and parse the event payload
payload = events.selectExpr("CAST(body AS STRING) AS payload")

Vous pouvez configurer des options Event Hubs supplémentaires telles que le groupe de consommateurs et la position de départ :

events = (spark.readStream
    .format("eventhubs")
    .option("eventhubs.connectionString", connection_string)
    .option("eventhubs.consumerGroup", "$Default")
    .option("eventhubs.startingPosition", '{"offset": "-1", "seqNo": -1, "enqueuedTime": null, "isInclusive": true}')
    .load())

L’option startingPosition contrôle l’emplacement où commencer la lecture des événements. Utilisez -1 pour les événements les plus récents ou configurez un décalage, un numéro de séquence ou un temps d’attente spécifique pour vos besoins d’ingestion.

Configurer les emplacements de point de contrôle

Les points de contrôle sont le mécanisme qui rend structured Streaming tolérant aux pannes. Le répertoire de checkpoint stocke les informations de décalage qui suivent exactement quels enregistrements ont été traités. Si votre travail de streaming s’arrête de façon inattendue, il peut redémarrer et continuer à partir du dernier point de contrôle validé.

Lorsque vous démarrez une écriture en streaming, spécifiez l’emplacement du point de contrôle :

checkpoint_path = "/checkpoints/sensor-ingestion"

query = (parsed_df.writeStream
    .format("delta")
    .option("checkpointLocation", checkpoint_path)
    .toTable("catalog.schema.sensor_readings"))

Stockez vos points de contrôle dans un emplacement fiable et durable comme Azure Data Lake Storage. Les fichiers de point de contrôle sont petits mais critiques pour la récupération. Configurez donc votre emplacement de stockage avec les stratégies de sauvegarde et de rétention appropriées.

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

L’écriture dans les tables Unity Catalogue utilise le format Delta Lake, qui prend en charge les écritures de streaming et les transactions ACID. L’API writeStream met en miroir l’API batch write , mais inclut des options pour le traitement continu :

(parsed_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("bronze.iot.sensor_readings"))

Le mode de sortie détermine la façon dont les résultats sont écrits :

• append : De nouvelles lignes sont ajoutées à la table. Utilisez ce mode pour la plupart des charges de travail d’ingestion.
• mise à jour : les lignes modifiées sont mises à jour. Utiliser avec des agrégations qui produisent des mises à jour.
• terminé : toutes les lignes sont réécrites. Utiliser avec des agrégations qui nécessitent un recalcul complet.

Pour les scénarios d’ingestion où vous ajoutez des données brutes, le mode par défaut append est approprié.

Contrôler le traitement avec des déclencheurs

Les déclencheurs déterminent quand et la fréquence à laquelle le moteur de streaming traite les données. Azure Databricks prend en charge plusieurs options de déclencheur :

Pour l’ingestion continue qui s’exécute indéfiniment, omettez l’option de déclencheur ou utilisez un intervalle de temps de traitement. Pour les modèles de lots incrémentiels dans lesquels vous souhaitez traiter toutes les données disponibles selon une planification, utilisez availableNow:

(parsed_df.writeStream
    .format("delta")
    .option("checkpointLocation", checkpoint_path)
    .trigger(availableNow=True)
    .toTable("catalog.schema.table_name"))

Le availableNow déclencheur traite toutes les données non traitées depuis le dernier point de contrôle, puis se termine. Vous pouvez planifier ce schéma à l'aide des Jobs Databricks pour s'exécuter à intervalles réguliers, en combinant l'efficacité du flux continu (uniquement le traitement de nouvelles données) avec la prévisibilité des travaux de traitement par lots planifiés.

Surveiller les requêtes en streaming

Les requêtes de diffusion en continu actives apparaissent dans l’interface utilisateur Spark sous l’onglet Streaming structuré . Vous pouvez également accéder par programme à l’état de la requête :

query = parsed_df.writeStream.format("delta").option("checkpointLocation", checkpoint_path).toTable("my_table")

# Check if the query is still running
print(query.isActive)

# Get the current status
print(query.status)

# View recent progress
print(query.recentProgress)

Utilisez la méthode pour bloquer l’exécution awaitTermination() jusqu’à ce que le flux se termine ou échoue. Pour les charges de travail de production, implémentez la gestion des erreurs pour capturer et répondre aux échecs de diffusion en continu :

try:
    query.awaitTermination()
except Exception as e:
    print(f"Stream failed: {e}")

Dans les environnements de production, configurez la surveillance et les alertes pour détecter quand les flux de diffusion en continu prennent du retard ou rencontrent des erreurs. Les métriques de streaming incluent des informations de décalage qui vous aident à comprendre si votre traitement suit le rythme des données entrantes.