リアルタイムモードのクエリパフォーマンスを最適化および監視する

このページでは、コンピュートのチューニング、エンドツーエンドの遅延を削減するためのテクニック、および突然モードでクエリのパフォーマンスを測定するためのアプローチについて説明します。

コンピュートチューニング

コンピュートを設定するときは、次の点を考慮してください。

• マイクロバッチモードとは異なり、リアルタイムタスクはデータ待ちの間アイドル状態になる可能性があるため、リソースの無駄遣いを避けるためには適切なサイジングが不可欠です。

• 次のように調整して、目標クラスター使用率レベル (50% など) を目指します。

  • maxPartitions （Kafka向け）
  • spark.sql.shuffle.partitions （シャッフルステージ用）

• Databricksは、オーバーヘッドを削減するために、各タスクが複数のKafkaパーティションを処理するようにmaxPartitionsを設定することを推奨しています。

• 単純な 1 段階のジョブのワークロードに合わせて、ワーカーごとのタスク スロットを調整します。

• シャッフルの多いジョブの場合は、バックログを回避するシャッフル パーティションの最小数を見つけて、そこから調整する経験をしてください。 十分なスロットがない場合、コンピュートはジョブをスケジュールしません。

注記

Databricks Runtime 16.4 LTS以降では、すべてのリアルタイムパイプラインがチェックポイントv2を使用し、リアルタイムモードとマイクロバッチモード間のシームレスな切り替えを可能にしています。

レイテンシー最適化

構造化ストリーミングのリアルタイムモードには、エンドツーエンドの遅延を低減するためのオプションの技術が用意されています。どちらもデフォルトでは有効になっていません。それぞれ個別に有効化する必要があります。

• 非同期進捗状況追跡：オフセットへの書き込みとコミットログを非同期スレッドに移動することで、ステートレスクエリのバッチ間時間を短縮します。
• 非同期状態チェックポイント：状態チェックポイントを待たずに、計算が完了するとすぐに次のマイクロバッチの処理を開始し、ステートフルクエリのレイテンシを削減します。

モニタリングと可観測性

起動モードでは、従来のバッチ期間のメトリクスは実際のエンドツーエンドのレイテンシを反映しません。 以下の方法を用いて、レイテンシを正確に測定し、クエリにおけるボトルネックを特定してください。

エンドツーエンドのレイテンシはワークロードによって異なり、ビジネスロジックを考慮に入れなければ正確に測定できない場合もあります。例えば、ソースタイムスタンプがKafkaに出力される場合、Kafkaの出力タイムスタンプとソースタイムスタンプの差としてレイテンシを計算できます。

組み込みのメトリクス StreamingQueryProgress

StreamingQueryProgressイベントはドライバログに自動的に記録され、 StreamingQueryListenerのonQueryProgress()コールバック関数を通じてアクセスできます。これにより、たとえばメトリクスを外部モニタリング システムに公開する場合など、進行状況イベントにプログラム的に対応できるようになります。 QueryProgressEvent.json()またはtoString()には、以下のリアルタイムモードのメトリクスが含まれます。

1. 処理レイテンシー (processingLatencyMs)。リアルタイムモードのクエリーがレコードを読み取ってから、そのクエリーが次のステージまたはダウンストリームに書き込むまでの経過時間。システムはこのメトリクスをタスクごとに報告します。
2. ソースキューイング遅延 （ sourceQueuingLatencyMs ）。システムがメッセージバスにレコードを書き込んだ時点（例えば、Kafkaのログ追加時間）から、リアルタイムモードのクエリが最初にそのレコードを読み取るまでの経過時間。システムはタスクごとにこのメトリクスを報告します。
3. エンドツーエンドの遅延 （ e2eLatencyMs ）。システムがレコードをメッセージバスに書き込んでから、リアルタイムモードのクエリがレコードを下流に書き込むまでの時間。システムは、すべてのタスクによって処理されたすべてのレコードにわたって、このメトリクスをバッチごとに集計します。

JSONの進捗イベントには、latencies の下にこれらのメトリクスが含まれます。例えば：

JSON

{
  "latencies": {
    "processingLatencyMs": {
      "P0": 0,
      "P50": 0,
      "P90": 0,
      "P95": 0,
      "P99": 0
    },
    "sourceQueuingLatencyMs": {
      "P0": 0,
      "P50": 1,
      "P90": 1,
      "P95": 2,
      "P99": 3
    },
    "e2eLatencyMs": {
      "P0": 0,
      "P50": 1,
      "P90": 1,
      "P95": 2,
      "P99": 4
    }
  }
}

タスクの使用率の監視

busyTimeFraction を使用して、Spark タスクの使用率が throughput を制限しているかどうかを判断します。1 に近い値は、タスクが完全に利用されており、クエリーにより多くのコンピュートが必要になる可能性があることを示しています。低い値は、タスクがアイドル状態かブロックされている時間が長いことを示しています。このメトリクスの範囲は 0 から 1 であり、ステージおよびタスクごとに報告されます。

ストリーミング クエリーを開始する前にデバッグ メトリクスを有効にします。

Python

spark.conf.set("spark.databricks.streaming.execution.enableDebugMetrics", "true")

次のいずれかの方法を使用してタスクメトリクスにアクセスします：

Raw Data

クエリーを起動または再開します。After a Trigger completes, open 生データ below the クエリー cell and find _taskMetrics under latencies.次の例は、タスクメトリクスを示しています。

JSON

{
  "latencies": {
    "_taskMetrics": {
      "stage_0_task_0": {
        "busyTimeFraction": 0.03
      }
    }
  }
}

Python

Triggerが完了したら、lastProgressを介して_taskMetricsにアクセスします：

Python

import json

task_metrics = json.loads(query.lastProgress.json)["latencies"]["_taskMetrics"]
print(task_metrics)

Observe API を使用したカスタムレイテンシ測定

Observe APIを使用すると、別のジョブを起動することなく、レイテンシをインラインで測定できます。ソースデータの到着時刻を近似するソースタイムスタンプがある場合は、シンクの前のタイムスタンプを記録し、その差を計算することで、バッチごとの遅延を推定できます。結果は進捗報告書に掲載され、リスナーが閲覧できる。

Python

Python

from datetime import datetime

from pyspark.sql.functions import avg, col, lit, max, percentile_approx, udf, unix_millis
from pyspark.sql.types import TimestampType

@udf(returnType=TimestampType())
def current_timestamp():
  return datetime.now()

# Query before outputting
.withColumn("temp-timestamp", current_timestamp())
.withColumn(
  "latency",
  unix_millis(col("temp-timestamp")).cast("long") - unix_millis(col("timestamp")).cast("long"))
.observe(
  "observedLatency",
  avg(col("latency")).alias("avg"),
  max(col("latency")).alias("max"),
  percentile_approx(col("latency"), lit(0.99), lit(150)).alias("p99"),
  percentile_approx(col("latency"), lit(0.5), lit(150)).alias("p50"))
.drop(col("latency"))
.drop(col("temp-timestamp"))
# Output part of the query. For example, .WriteStream, etc.

Scala

Scala

import org.apache.spark.sql.functions.{avg, col, lit, max, percentile_approx, udf, unix_millis}

val currentTimestampUDF = udf(() => System.currentTimeMillis())

// Query before outputting
.withColumn("temp-timestamp", currentTimestampUDF())
.withColumn(
  "latency",
  col("temp-timestamp").cast("long") - unix_millis(col("timestamp")).cast("long"))
.observe(
  name = "observedLatency",
  avg(col("latency")).as("avg"),
  max(col("latency")).as("max"),
  percentile_approx(col("latency"), lit(0.99), lit(150)).as("p99"),
  percentile_approx(col("latency"), lit(0.5), lit(150)).as("p50"))
.drop(col("latency"))
.drop(col("temp-timestamp"))
// Output part of the query. For example, .WriteStream, etc.

サンプル出力：

JSON

"observedMetrics" : {
  "observedLatency" : {
    "avg" : 63.8369765176552,
    "max" : 219,
    "p99" : 154,
    "p50" : 49
  }
}

その他のリソース

• リアルタイムモード参照