Spark 構造化ストリーミングを使用してデータを取り込む

リアルタイム データ インジェストを使用すると、バッチ間隔を待機するのではなく、発生したイベントを処理できます。 組織で IoT センサーの読み取り値、クリックストリームデータ、または財務トランザクションをリアルタイムで取得する必要がある場合、Spark Structured Streaming はレイクハウスへの継続的なデータ取り込みの基盤を提供します。

このユニットでは、ストリーミング ソースから読み取り、Unity カタログ テーブルに書き込む構造化ストリーミング ジョブを構成する方法について説明します。

構造化ストリーミング モデルを理解する

構造化ストリーミングでは、受信データは、新しいデータが到着すると継続的に拡張される無制限のテーブルとして扱われます。 既に知っているのと同じ DataFrame API を使用しますが、 spark.readではなく、 spark.readStream を使用して、データを段階的に処理するストリーミング DataFrame を作成します。

センサー データが製造機器から継続的に到着するシナリオを考えてみましょう。 このデータを取り込むための 1 時間ごとのバッチ ジョブをスケジュールするのではなく、受信から数秒以内に各読み取りを処理するストリーミング ジョブを構成できます。 ストリーミング エンジンは 、チェックポイントを使用して処理されたレコードを追跡するため、ジョブが再起動すると、データを再処理したり失ったりすることなく、中断した場所から再開されます。

ストリーミング インジェストの基本的なパターンには、次の 3 つの手順が含まれます。

1. 接続の詳細を含むを使用してspark.readStream
2. 必要に応じて変換を適用 してデータを整形する
3. チェックポイントの場所を設定してwriteStream

Kafka と Event Hubs に接続する

Azure Event Hubs には Apache Kafka と互換性のあるエンドポイントが用意されています。つまり、両方のサービスに同じ Kafka コネクタを使用できます。 この統合されたアプローチにより、さまざまなメッセージ バスを操作する際のコードが簡略化されます。

次の例では、Kafka トピックからメッセージを読み取ります。

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

Azure Event Hubs の場合は、Kafka と互換性のあるインターフェイスを介して接続します。 主な違いは、Event Hubs 接続文字列で SASL を使用する認証構成です。

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())

Kafka 互換インターフェイスを介して Kafka または Event Hubs から読み取るときに、 key、 value、 topic、 partition、 offset、および timestamp 列を含む標準スキーマを受け取ります。 key列とvalue列にはバイナリ データが含まれているため、通常は文字列にキャストしてコンテンツを解析します。

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.*"))

ネイティブ Event Hubs コネクタを使用する

Azure Databricks には、認証を簡略化し、Event Hubs 固有の機能に直接アクセスできるネイティブ Event Hubs コネクタも用意されています。 このコネクタでは、 eventhubs 形式が使用されます。

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

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

ネイティブ Event Hubs コネクタは、Kafka 互換インターフェイスとは異なるスキーマを返します。 イベント ペイロードは、追加の Event Hubs 固有のメタデータ列と共に(bodyではなく) value列に格納されます。

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

コンシューマー グループや開始位置などの追加の Event Hubs オプションを構成できます。

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())

startingPosition オプションは、イベントの読み取りを開始する場所を制御します。 最新のイベントに -1 を使用するか、インジェストのニーズに合わせて特定のオフセット、シーケンス番号、またはエンキュー時間を構成します。

チェックポイントの場所を構成する

チェックポイントは、構造化ストリーミングのフォールト トレラントを実現するメカニズムです。 チェックポイント ディレクトリには、処理されたレコードを正確に追跡するオフセット情報が格納されます。 ストリーミング ジョブが予期せず停止した場合は、最後にコミットされたチェックポイントから再開して続行できます。

ストリーミング書き込みを開始するときに、チェックポイントの場所を指定します。

checkpoint_path = "/checkpoints/sensor-ingestion"

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

Azure Data Lake Storage などの信頼性の高い永続的な場所にチェックポイントを格納します。 チェックポイント ファイルは小さいが回復には重要であるため、適切なバックアップと保持ポリシーを使用してストレージの場所を構成します。

ストリーミング データを Unity カタログ テーブルに書き込む

Unity カタログ テーブルへの書き込みでは、ストリーミング書き込みと ACID トランザクションの両方をサポートする Delta Lake 形式が使用されます。 writeStream API はバッチ write API を反映しますが、継続的な処理のオプションが含まれています。

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

出力モードは、結果の書き込み方法を決定します。

• append: 新しい行がテーブルに追加されます。 このモードは、ほとんどのインジェスト ワークロードに使用します。
• update: 変更された行が更新されます。 更新を生成する集計で使用します。
• complete: すべての行が書き換えられます。 完全な再計算を必要とする集計で使用します。

生データを追加するインジェスト シナリオでは、既定の append モードが適しています。

トリガーを使用した処理の制御

トリガーは、ストリーミング エンジンがデータを処理するタイミングと頻度を決定します。 Azure Databricks では、いくつかのトリガー オプションがサポートされています。

無期限に実行される継続的インジェストの場合は、トリガー オプションを省略するか、処理時間間隔を使用します。 利用可能なすべてのデータをスケジュールに従って処理する増分バッチ パターンの場合は、次の availableNowを使用します。

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

availableNow トリガーは、最後のチェックポイント以降に未処理のすべてのデータを処理してから終了します。 このパターンをスケジュールするには、Databricks ジョブを使用して一定の間隔で実行し、ストリーミングの効率 (新しいデータの処理のみ) とスケジュールされたバッチ ジョブの予測可能性を組み合わせて使用できます。

ストリーミング クエリの監視

アクティブなストリーミング クエリは、Spark UI の [ 構造化ストリーミング ] タブに表示されます。プログラムでクエリの状態にアクセスすることもできます。

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)

ストリームが完了するか失敗するまで実行をブロックするには、 awaitTermination() メソッドを使用します。 運用環境のワークロードの場合は、エラー処理を実装して、ストリーミング エラーをキャプチャして対応します。

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

運用環境では、ストリーミング ジョブが遅れたりエラーが発生したりするタイミングを検出するように監視とアラートを構成します。 ストリーミング メトリックには、処理が受信データに追い付いているかどうかを理解するのに役立つラグ情報が含まれます。