リアルタイム モードを設定する

このページでは、Structured Streaming でリアルタイム モード クエリを実行するために必要な前提条件と構成について説明します。 詳細なチュートリアルについては、「 チュートリアル: リアルタイム ストリーミング ワークロードを実行する」を参照してください。 リアルタイム モードの概念については、「 構造化ストリーミングのリアルタイム モード」を参照してください。

前提 条件

リアルタイム モードを使用するには、次の要件を満たすようにコンピューティングを構成する必要があります。

• クラシック コンピューティングを使用します。 専用および標準のアクセス モードがサポートされています。 標準アクセス モードは、Pythonでのみサポートされています。 Lakeflow Spark 宣言型パイプラインとサーバーレス クラスターはサポートされていません。
• Databricks Runtime 16.4 LTS 以降を使用します。
• 自動スケールをオフにします。
• Photon をオフにします。
• spark.databricks.streaming.realTimeMode.enabled を true に設定します。
• 中断を回避するためにスポット インスタンスをオフにします。

UDF を使用した待機時間の影響を受けやすいワークロードの場合、Databricks では専用アクセス モードを使用することをお勧めします。 テーブル関数を参照してください。

クラシック コンピューティングの作成と構成の手順については、「 コンピューティング構成リファレンス」を参照してください。

Stream to stream 結合

ストリームからストリームへの内部結合には、リアルタイム モードの追加構成が必要です。 外部結合はサポートされていません。 「Stream to stream 結合」を参照してください。

リアルタイム モードでストリーム間の結合を有効にするには、次の Spark 構成を設定します。

Python

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

Scala

spark.conf.set("spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled", "true")
spark.conf.set("spark.sql.streaming.join.stateFormatVersion", "4")
spark.conf.set("spark.sql.streaming.join.stateFormatV4.enabled", "true")
spark.conf.set("spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion", "2")
spark.conf.set("spark.sql.streaming.realTimeMode.controlMessage.enabled", "true")

SQL

SET spark.databricks.streaming.realTimeMode.streamStreamJoin.enabled = true;
SET spark.sql.streaming.join.stateFormatVersion = 4;
SET spark.sql.streaming.join.stateFormatV4.enabled = true;
SET spark.sql.streaming.stateStore.rocksdb.mergeOperatorVersion = 2;
SET spark.sql.streaming.realTimeMode.controlMessage.enabled = true;

クエリの構成

リアルタイム モードでクエリを実行するには、リアルタイム トリガーを有効にする必要があります。 リアルタイム トリガーは、更新モードでのみサポートされます。

Python

query = (
    spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("subscribe", input_topic)
        .load()
        .writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", broker_address)
        .option("topic", output_topic)
        .option("checkpointLocation", checkpoint_location)
        .outputMode("update")
        # In PySpark, the realTime trigger requires specifying the interval.
        .trigger(realTime="5 minutes")
        .start()
)

Scala

import org.apache.spark.sql.execution.streaming.RealTimeTrigger

val readStream = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("subscribe", inputTopic).load()
      .writeStream
      .format("kafka")
      .option("kafka.bootstrap.servers", brokerAddress)
      .option("topic", outputTopic)
      .option("checkpointLocation", checkpointLocation)
      .outputMode("update")
      .trigger(RealTimeTrigger.apply())
      // RealTimeTrigger can also accept an argument specifying the checkpoint interval.
      // For example, this code indicates a checkpoint interval of 5 minutes:
      // .trigger(RealTimeTrigger.apply("5 minutes"))
      .start()

コンピューティングのサイズ設定

コンピューティングに十分なタスク スロットがある場合は、コンピューティング リソースごとに 1 つのリアルタイム ジョブを実行できます。

低待機時間モードで実行するには、使用可能なタスク スロットの合計数が、すべてのクエリ ステージのタスク数以上である必要があります。

スロット計算の例

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

その他のリソース

• チュートリアル: リアルタイム ストリーミング ワークロードを実行する
• リアルタイム モードの例
• リアルタイム モードのクエリ パフォーマンスを最適化および監視する
• リアルタイム モードリファレンス