Lakeflow パイプラインでリアルタイム モードを使用する

リアルタイム モードでは、超低待機時間のデータ処理が可能になり、エンド ツー エンドの待機時間は 5 ミリ秒と短くなります。 不正行為の検出やリアルタイムのパーソナル化など、ストリーミング データへの即時対応を必要とする運用ワークロードには、リアルタイム モードを使用します。

リアルタイム モードは、パイプラインの外部の構造化ストリーミングでも直接使用できます。 構造化ストリーミングのリアルタイム モードを参照してください。

リアルタイム モードで低待機時間を実現する方法

リアルタイム モードは、次の 3 つの重要な方法で標準の連続処理とは異なります。

• 実行時間の長いバッチ: システムは、実行時間の長いバッチ内でソースで使用できるようになるとデータを処理します (既定値は 5 分です)。
• 同時ステージ スケジューリング: すべてのクエリ ステージが同時にスケジュールされます。 コンピューティング リソースには、すべてのステージを同時にカバーするのに十分な使用可能なタスク スロットが必要です。 「 コンピューティングのサイズ設定」を参照してください。
• ストリーミング シャッフル: データは、ダウンストリーム ステージを開始する前にアップストリーム ステージが完了するのを待つのではなく、生成されるとすぐにステージ間で渡されます。

チェックポイント間隔 ( pipelines.trigger.interval を使用して構成) は、永続的ストレージに状態とソースのオフセットを保持する頻度を制御します。 間隔が長いほどチェックポイント処理のオーバーヘッドは軽減されますが、失敗後の復旧時間が長くなり、メトリックの報告が遅れます。 間隔を短くすると持続性が向上しますが、オーバーヘッドが増加します。

リアルタイム モードと継続的パイプライン

リアルタイム モードは、特殊な種類の連続トリガーです。 連続モードは引き続き必要です。リアルタイム モードでは、フロー レベルの待機時間の最適化が上に追加されます。 リアルタイム モードを使用するには、パイプラインを最初に連続モードで実行する必要があります。 その後、リアルタイム モードでは、フロー レベルで追加の最適化が適用され、標準の連続処理を超える 2 秒未満の待機時間が実現されます。

リアルタイム モードを有効にするには、次の 3 つの構成手順が必要です。

1. パイプラインを連続モードに設定します。
2. パイプライン レベルでリアルタイム モードを有効にします。
3. リアルタイムの更新フローを定義します。

Requirements

リアルタイム モードを構成する

手順 1: パイプラインを連続モードに設定する

パイプライン設定で、[ パイプライン モード ] を [継続的] に設定するか、パイプライン JSON で設定します。

{
  "continuous": true
}

手順 2: パイプライン レベルでリアルタイム モードを有効にする

パイプラインの設定で、[ 詳細] > Spark 構成の Spark 構成に次のキーを追加します。

spark.databricks.streaming.realTimeMode.enabled = true

これは、パイプライン JSON で設定することもできます。

{
  "continuous": true,
  "spark_conf": {
    "spark.databricks.streaming.realTimeMode.enabled": "true"
  }
}

手順 3: リアルタイム更新フローを定義する

リアルタイム モードでは、更新フローが必要です。 dp.create_sink()を使用して出力先を定義し、次に@dp.update_flowデコレーターを使用して、pipelines.triggerを"RealTime"に設定し、targetでシンクを指定します。

from pyspark import pipelines as dp

# Define the output sink
dp.create_sink(
    "my_kafka_sink",
    "kafka",
    {
        "kafka.bootstrap.servers": "<bootstrap-servers>",
        "topic": "<output-topic>",
    }
)

# Define the real-time update flow targeting the sink
@dp.update_flow(
    name="my_rtm_flow",
    target="my_kafka_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",  # optional; defaults to 5 minutes
    }
)
def my_real_time_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", "<bootstrap-servers>")
            .option("subscribe", "<input-topic>")
            .load()
    )

フロー レベルの構成パラメーター:

コード例

Kafka から Kafka へ

Kafka トピックから読み取り、Kafka 出力ターゲットに書き込みます。

from pyspark import pipelines as dp

dp.create_sink("kafka_output_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="kafka_rtm_flow",
    target="kafka_output_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def kafka_rtm_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .option("startingOffsets", "latest")
            .load()
            .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "timestamp")
    )

ブロードキャスト結合を使用したエンリッチ

静的参照テーブルに対して Kafka ストリームを結合します。 ブロードキャスト (ストリームから静的) の結合のみがサポートされます。 ストリーム間結合は、リアルタイム モードではサポートされていません。

from pyspark import pipelines as dp
from pyspark.sql.functions import broadcast, expr

dp.create_sink("enriched_output_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": enriched_output_topic,
})

@dp.update_flow(
    name="enriched_events_flow",
    target="enriched_output_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def enriched_events():
    lookup = spark.read.table("catalog.schema.lookup_table")
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
            .withColumn("event_key", expr("CAST(value AS STRING)"))
            .join(broadcast(lookup), expr("event_key = lookup_key"))
            .select("event_key", "lookup_value", "timestamp")
    )

集計

ステートフル groupByを使用して、キーによってイベントをカウントします。 ステートフル操作の入力パーティション数と一致するように spark.sql.shuffle.partitions を設定します。

from pyspark import pipelines as dp
from pyspark.sql.functions import col

dp.create_sink("event_counts_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="event_counts_flow",
    target="event_counts_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
        "spark.sql.shuffle.partitions": "8",
    }
)
def event_counts():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
            .selectExpr("CAST(key AS STRING) AS event_type", "timestamp")
            .groupBy(col("event_type"))
            .count()
    )

サポートされているソースとシンク

計算リソースのサイズ設定

コンピューティングに十分なタスク スロットがある場合は、コンピューティング リソースごとに 1 つのリアルタイム パイプラインを実行できます。 使用可能なタスク スロットは、すべてのクエリ ステージのすべてのタスクをカバーする必要があります。

maxPartitions設定しない場合は、Kafka トピックのパーティションの数を使用します。

オペレーターのサポート

transformWithState リアルタイムモード

transformWithState は、マイクロバッチ処理とは次の違いがあり、リアルタイム モードでサポートされています。

• handleInputRows は、バッチごとにキーごとに 1 回ではなく、行ごとに 1 回呼び出されます。 inputRows反復子は、呼び出しごとに 1 つの値を生成します。
• イベント時間タイマーはサポートされていません。 処理時間タイマーは、データが到着していない場合に実行時間の長いバッチが終了したときに発生します。
• transformWithStateInPandas はサポートされません。

リアルタイム モードでの Pandas UDF

pandas UDF で待機時間を最小限に抑えるには、 spark.sql.execution.arrow.maxRecordsPerBatch を 1 に設定します。 これにより、スループットを犠牲にして待機時間が最適化されます。 スループットも重要な場合は、この値を 100 以上に設定します。

リアルタイム モードのパフォーマンスを監視する

リアルタイム モードでは、StreamingQueryProgress フィールドのlatenciesで待機時間メトリックが公開されます。 これらのメトリックには、 StreamingQueryListener 経由でアクセスするか、ストリーミング クエリの lastProgress プロパティを調べます。

各メトリックは、p50、p90、p95、および p99 パーセンタイルとして報告されます。

制限事項

パイプラインごとに 1 つのリアルタイム フローを使用することをお勧めします。 複数のフローが許可されますが、フロー間のタスク スロットの競合により待機時間が長くなります。

演算子とソースの制限事項の完全な一覧については、 リアルタイム モードの制限事項を参照してください。

その他のリソース

• リアルタイムモードの概念
• リアルタイム モードリファレンス
• トリガー型パイプライン モードと連続パイプライン モード
• パイプラインのクラシック コンピューティングを構成する