Optymalizowanie i monitorowanie wydajności zapytań w trybie rzeczywistym

Na tej stronie opisano dostrajanie zasobów obliczeniowych, techniki zmniejszania całkowitego opóźnienia oraz podejścia do mierzenia wydajności zapytań w trybie czasu rzeczywistego.

Dostrajanie zasobów obliczeniowych

Podczas konfigurowania obliczeń należy wziąć pod uwagę następujące kwestie:

  • W przeciwieństwie do trybu mikrosadowego, zadania w czasie rzeczywistym mogą pozostawać bezczynne podczas oczekiwania na dane, więc odpowiednie dopasowanie rozmiaru jest niezbędne, aby uniknąć marnowania zasobów.
  • Celuj w celu osiągnięcia docelowego poziomu wykorzystania klastra, takiego jak 50%, przez dostrajanie:
    • maxPartitions (dla platformy Kafka)
    • spark.sql.shuffle.partitions (dla etapów tasowania)
  • Usługa Databricks zaleca ustawienie maxPartitions tak, aby każde zadanie obsługiwało wiele partycji platformy Kafka w celu zmniejszenia obciążenia.
  • Dostosuj gniazda zadań przypadające na pracownika, aby dopasować obciążenie pracą do prostych zadań jednofazowych.
  • W przypadku zadań wymagających intensywnego mieszania eksperymentuj, aby znaleźć minimalną liczbę partycji mieszania, które unikają opóźnień, a następnie dostosuj wartość początkową. Obliczenia nie będą planować zadania, jeśli nie ma wystarczającej liczby miejsc.

Uwaga / Notatka

W środowisku Databricks Runtime 16.4 LTS i nowszym wszystkie rurki w czasie rzeczywistym używają punktów kontrolnych w wersji 2, aby umożliwić płynne przełączanie z trybów czasu rzeczywistego na mikropartii.

Optymalizacja opóźnienia

Począwszy od Databricks Runtime 18 LTS, tryb czasu rzeczywistego Structured Streaming domyślnie włącza następujące techniki w celu zmniejszenia opóźnień end-to-end, więc nie musisz ich konfigurować:

  • Śledzenie postępu asynchronicznego: przenosi zapisy do dzienników przesunięcia i zatwierdzania do wątku asynchronicznego, co zmniejsza czas międzysadowy dla zapytań bezstanowych.
  • Asynchroniczne tworzenie punktów kontrolnych stanu: rozpoczyna przetwarzanie następnej mikropartii natychmiast po zakończeniu obliczeń, bez potrzeby oczekiwania na tworzenie punktów kontrolnych stanu, co zmniejsza opóźnienie zapytań stanowych.

Jeśli wyraźnie ustawisz opcję asyncProgressTrackingEnabled lub konfigurację spark.databricks.streaming.statefulOperator.asyncCheckpoint.enabled, twoje ustawienie ma pierwszeństwo.

Monitorowanie i obserwowanie

W trybie czasu rzeczywistego tradycyjne metryki czasu trwania partii nie odzwierciedlają rzeczywistego opóźnienia kompleksowego. Skorzystaj z poniższych metod, aby dokładnie zmierzyć opóźnienie i zidentyfikować wąskie gardła w zapytaniach.

Kompleksowe opóźnienie jest specyficzne dla obciążenia, a czasami może być dokładnie mierzone tylko za pomocą logiki biznesowej. Jeśli na przykład źródłowy znacznik czasu jest przesyłany w Kafka, możesz obliczyć opóźnienie jako różnicę między sygnaturą czasową danych wyjściowych w Kafka a sygnaturą czasową źródła.

Wbudowane metryki z StreamingQueryProgress

Zdarzenie StreamingQueryProgress jest automatycznie rejestrowane w dziennikach sterowników i dostępne za pośrednictwem funkcji wywołania zwrotnego StreamingQueryListener.onQueryProgress() Dzięki temu można programowo reagować na zdarzenia postępu, na przykład jeśli chcesz opublikować metryki w zewnętrznym systemie monitorowania. QueryProgressEvent.json() lub toString() obejmują te metryki trybu w czasie rzeczywistym:

  1. Opóźnienie przetwarzania (processingLatencyMs): Czas od momentu, gdy zapytanie w trybie rzeczywistym po raz pierwszy odczytuje rekord, do momentu, gdy zacznie zapisywać go do sink. Obejmuje to wszystkie etapy przetwarzania, ale nie uwzględnia czasu poświęconego na zapisywanie do odbiornika.
  2. Opóźnienie kolejki źródłowej (sourceQueuingLatencyMs): Czas od momentu, gdy system zapisuje rekord na magistrali wiadomości, na przykład czas dodawania logów w Kafce, do momentu, gdy zapytanie w trybie rzeczywistym po raz pierwszy odczyta rekord.
  3. Opóźnienie end-to-end (e2eLatencyMs): Czas od momentu, gdy system zapisuje rekord na magistrali wiadomości, do momentu, gdy zapytanie w trybie rzeczywistym zaczyna zapisywać go do sinku. To nie uwzględnia czasu spędzanego na pisaniu do ujścia.

Spark agreguje te wskaźniki dla każdej partii we wszystkich przetworzonych rekordach i raportuje je jako rozkłady percentylowe. Zdarzenie postępu JSON zawiera następujące metryki w elemencie latencies. Przykład:

{
  "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
    }
  }
}

Monitorowanie wykorzystania zadań

Użyj busyTimeFraction do określenia, czy wykorzystanie zadań Spark ogranicza przepustowość. Wartości bliskie 1 wskazują, że zadania są w pełni wykorzystane i zapytanie może wymagać więcej obciążeń obliczeniowych. Niższe wartości wskazują, że zadania spędzają więcej czasu w trybie nieczynnym lub zablokowanym. Metryka ta mieści się w zakresie od 0 do 1 i jest raportowana na każdy etap oraz zadanie.

Włącz metryki debugowania przed rozpoczęciem zapytania streamingowego:

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

Użyj jednej z następujących metod dostępu do metryk zadań:

Surowe dane

Rozpocznij lub uruchom zapytanie od nowa. Po zakończeniu działania wyzwalacza otwórz Surowe dane poniżej komórki zapytania i znajdź _taskMetrics w sekcji latencies. Poniższy przykład przedstawia metryki zadania:

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

Python

Po zakończeniu działania wyzwalacza uzyskaj dostęp do _taskMetrics za pomocą lastProgress:

import json

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

Niestandardowy pomiar latencji za pomocą API Observe

Interfejs API Observer umożliwia pomiar opóźnienia na miejscu bez uruchamiania oddzielnego zadania. Jeśli masz znacznik czasu źródła, który przybliża czas przybycia danych źródłowych, możesz oszacować opóźnienie na partię, rejestrując znacznik czasu przed miejscem docelowym i obliczając różnicę. Wyniki są wyświetlane w raportach postępów i są dostępne dla słuchaczy.

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 - język programowania

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.

Przykładowe dane wyjściowe:

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

Dodatkowe zasoby

  • Ostatnia aktualizacja o