Оптимизация и мониторинг производительности запросов в режиме реального времени

На этой странице рассматриваются параметры вычислений, методы сокращения сквозной задержки и подходы к измерению производительности запросов в режиме реального времени.

Настройка вычислений

При настройке вычислений рассмотрите следующее:

  • В отличие от микробатчевого режима, задачи в режиме реального времени могут оставаться в состоянии простоя во время ожидания данных, поэтому важно правильно распределять ресурсы, чтобы избежать их потерь.
  • Достичь уровня использования кластера, например 50%, путем настройки:
    • maxPartitions (для Kafka)
    • spark.sql.shuffle.partitions (для этапов перетасовки)
  • Databricks рекомендует задать параметр maxPartitions , чтобы каждая задача обрабатывала несколько секций Kafka, чтобы снизить затраты.
  • Настройте слоты задач для каждого рабочего в соответствии с рабочей нагрузкой для простых одноэтапных задач.
  • Для задач с интенсивным использованием ресемплирования проведите эксперименты, чтобы определить минимальное количество секций ресемплирования, которое позволит избежать задержек, и корректируйте настройки оттуда. Вычислительный узел не запланирует задание, если у него недостаточно слотов.

Замечание

В Databricks Runtime 16.4 LTS и более поздних версиях все конвейеры в режиме реального времени используют контрольные точки v2 для простого перехода между режимами реального времени и микропакетного режима.

Оптимизация задержки

Структурированный режим потоковой передачи в режиме реального времени имеет необязательные методы снижения сквозной задержки. По умолчанию ни одна из этих функций не включена. Их необходимо включить отдельно.

Мониторинг и наблюдаемость

В режиме реального времени традиционные метрики продолжительности пакета не отражают фактическую сквозную задержку. Используйте приведенные ниже подходы, чтобы точно определить задержку и определить узкие места в запросах.

Сквозная задержка по времени является специфичной для конкретной рабочей нагрузки и её точное измерение иногда возможно только с учётом бизнес-логики. Например, если исходная метка времени выводится в Kafka, можно вычислить задержку в качестве разницы между меткой времени вывода Kafka и исходной меткой времени.

Встроенные метрики с StreamingQueryProgress

Событие StreamingQueryProgress автоматически регистрируется в журналах драйвера и доступно через функцию обратного вызова StreamingQueryListener и onQueryProgress(). Это позволяет реагировать на события хода выполнения программным способом, например, если вы хотите опубликовать метрики во внешней системе мониторинга. QueryProgressEvent.json() или toString() включите следующие метрики режима реального времени:

  1. Задержка обработки (processingLatencyMs). Время, затраченное между тем, когда запрос в режиме реального времени считывает запись и когда запрос записывает его на следующий этап или ниже. Система сообщает эту метрику для каждой задачи.
  2. Задержка очереди источника (sourceQueuingLatencyMs). Время, прошедшее с момента записи системой записи в шину сообщений (например, время добавления журнала в Kafka) до момента, когда запрос в режиме реального времени впервые считывает запись. Система сообщает эту метрику для каждой задачи.
  3. Сквозная задержка (e2eLatencyMs). Время, когда система записывает запись в шину сообщений и когда запрос в режиме реального времени записывает запись вниз. Система агрегирует эту метрику на пакетной основе во всех записях, обработанных всеми задачами.

Событие прогресса JSON включает эти метрики под latencies. Рассмотрим пример.

{
  "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 пропускную способность. Значения около 1 указывают на то, что задачи полностью используются, и запросу может потребоваться больше вычислений. Низкие значения указывают на то, что задачи больше времени проводят в простое или заблокированы. Метрика варьируется от 0 до 1 и отражается по этапам и задачам.

Включите метрики отладки перед началом потокового запроса:

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

Используйте один из следующих методов для доступа к метрикам задач:

Необработанные данные

Запустите или перезапустите запрос. После завершения триггера откройте Raw Data под ячейкой запроса и найдите _taskMetrics под latencies. Следующий пример показывает метрики задачи:

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

Python

После завершения триггера доступны _taskMetrics–lastProgress:

import json

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

Настраиваемое измерение задержки с помощью API наблюдения

API наблюдения позволяет измерять задержку встроенным образом, не запуская отдельное задание. Если у вас есть метка времени источника, которая приближает время прибытия исходных данных, вы можете оценить задержку на каждом пакете, записав метку времени перед узлом приема и вычислив разницу. Результаты отображаются в отчетах о ходе выполнения и доступны прослушивателям.

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

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.

Образец вывода:

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

Дополнительные ресурсы

  • Last updated on