Behandeln großer Abfragen in interaktiven Workflows

Eine Herausforderung bei interaktiven Datenworkflows besteht darin, große Abfragen zu verarbeiten. Dazu gehören Abfragen, die zu viele Ausgabezeilen generieren, viele externe Partitionen abrufen oder auf extrem großen Datasets berechnen. Diese Abfragen können extrem langsam sein, die Computerressourcen auslasten und es erschweren, dass andere dieselben Ressourcen gemeinsam nutzen.

Query Watchdog ist ein Prozess, der verhindert, dass Abfragen Rechnerressourcen monopolisieren, durch die Untersuchung der häufigsten Ursachen großer Abfragen und beendet Abfragen, die einen Schwellenwert überschreiten. In diesem Artikel wird beschrieben, wie Sie Query Watchdog aktivieren und konfigurieren.

Von Bedeutung

Die Abfrageüberwachung ist für alle zweckbezogenen Berechnungen aktiviert, die mit der Benutzeroberfläche erstellt wurden.

Beispiel für eine störende Abfrage

Ein Analyst führt einige Ad-hoc-Abfragen in einem Just-in-Time-Data Warehouse durch. Der Analyst verwendet eine gemeinsam genutzte Berechnung zur automatischen Skalierung, die es mehreren Benutzern erleichtert, gleichzeitig eine einzelne Berechnung zu verwenden. Angenommen, es gibt zwei Tabellen, die jeweils eine Million Zeilen enthalten.

import org.apache.spark.sql.functions._
spark.conf.set("spark.sql.shuffle.partitions", 10)

spark.range(1000000)
  .withColumn("join_key", lit(" "))
  .createOrReplaceTempView("table_x")
spark.range(1000000)
  .withColumn("join_key", lit(" "))
  .createOrReplaceTempView("table_y")

Diese Tabellengrößen sind in Apache Spark verwaltbar. Sie enthalten jedoch jeweils eine join_key Spalte mit einer leeren Zeichenfolge in jeder Zeile. Dies kann passieren, wenn die Daten nicht perfekt sauber sind oder wenn es erhebliche Datenverzerrung gibt, bei denen bestimmte Schlüssel häufiger auftreten als andere. Diese leeren Verknüpfungsschlüssel sind weit verbreiteter als jeder andere Wert.

Im folgenden Code fügt der Analyst diese beiden Tabellen zu ihren Schlüsseln hinzu, was zu einer Ausgabe von einer Billion Ergebnissen führt, und alle diese werden auf einem einzigen Executor (dem Executor, der den " " Schlüssel erhält) erstellt:

SELECT
  id, count(id)
FROM
  (SELECT
    x.id
  FROM
    table_x x
  JOIN
    table_y y
  on x.join_key = y.join_key)
GROUP BY id

Diese Abfrage scheint zu laufen. Aber ohne über die Daten zu wissen, sieht der Analyst, dass "nur" eine einzelne Aufgabe im Laufe der Ausführung des Auftrags übrig bleibt. Die Abfrage ist nie abgeschlossen, sodass der Analyst frustriert und verwirrt darüber ist, warum es nicht funktioniert hat.

In diesem Fall gibt es nur einen problematischen Verknüpfungsschlüssel. Manchmal kann es viel mehr geben.

Aktivieren und Konfigurieren von Query Watchdog

Zum Aktivieren und Konfigurieren von Query Watchdog sind die folgenden Schritte erforderlich.

• Aktivieren Sie Watchdog mit spark.databricks.queryWatchdog.enabled.
• Konfigurieren Sie die Aufgabenlaufzeit mit spark.databricks.queryWatchdog.minTimeSecs.
• Ausgabe anzeigen mit spark.databricks.queryWatchdog.minOutputRows.
• Konfigurieren Des Ausgabeverhältnisses mit spark.databricks.queryWatchdog.outputRatioThreshold.

Um zu verhindern, dass eine Abfrage zu viele Ausgabezeilen für die Anzahl der Eingabezeilen erstellt, können Sie Query Watchdog aktivieren und die maximale Anzahl von Ausgabezeilen als ein Vielfaches der Anzahl der Eingabezeilen konfigurieren. In diesem Beispiel verwenden wir ein Verhältnis von 1000 (standard).

spark.conf.set("spark.databricks.queryWatchdog.enabled", true)
spark.conf.set("spark.databricks.queryWatchdog.outputRatioThreshold", 1000L)

Die letztere Konfiguration deklariert, dass jede bestimmte Aufgabe niemals mehr als 1000 Mal die Anzahl der Eingabezeilen erzeugen sollte.

Tipp

Das Ausgabeverhältnis ist vollständig anpassbar. Wir empfehlen, niedriger zu beginnen und zu sehen, welche Schwelle für Sie und Ihr Team gut funktioniert. Ein Bereich von 1.000 bis 10.000 ist ein guter Ausgangspunkt.

Nicht nur verhindert Query Watchdog, dass Benutzer Computeressourcen für Aufträge monopolisieren, die nie abgeschlossen werden, es spart auch Zeit, indem eine Abfrage, die nie abgeschlossen wäre, schnell fehlschlägt. Die folgende Abfrage schlägt beispielsweise nach mehreren Minuten fehl, da sie das Verhältnis überschreitet.

SELECT
  z.id
  join_key,
  sum(z.id),
  count(z.id)
FROM
  (SELECT
    x.id,
    y.join_key
  FROM
    table_x x
  JOIN
    table_y y
  on x.join_key = y.join_key) z
GROUP BY join_key, z.id

Hier sehen Sie Folgendes:

Normalerweise reicht es aus, die Abfrageüberwachung zu aktivieren und das Ausgabe-/Eingabeschwellenverhältnis festzulegen, aber Sie haben auch die Möglichkeit, zwei zusätzliche Eigenschaften festzulegen: spark.databricks.queryWatchdog.minTimeSecs und spark.databricks.queryWatchdog.minOutputRows. Diese Eigenschaften geben die Minimale Zeit an, die eine bestimmte Aufgabe in einer Abfrage ausführen muss, bevor sie abgebrochen wird, und die mindeste Anzahl der Ausgabezeilen für einen Vorgang in dieser Abfrage.

Sie können z. B. auf einen höheren Wert festlegen minTimeSecs , wenn Sie ihm die Möglichkeit geben möchten, eine große Anzahl von Zeilen pro Aufgabe zu erzeugen. Ebenso können Sie auf zehn Millionen festlegen spark.databricks.queryWatchdog.minOutputRows , wenn Sie eine Abfrage nur beenden möchten, nachdem eine Aufgabe in dieser Abfrage zehn Millionen Zeilen erstellt hat. Alles weniger und die Abfrage ist erfolgreich, auch wenn das Ausgabe-/Eingabeverhältnis überschritten wurde.

spark.conf.set("spark.databricks.queryWatchdog.minTimeSecs", 10L)
spark.conf.set("spark.databricks.queryWatchdog.minOutputRows", 100000L)

Tipp

Wenn Sie Query Watchdog in einem Notizbuch konfigurieren, wird die Konfiguration nicht über Computeneustarts hinweg beibehalten. Wenn Sie Query Watchdog für alle Benutzer einer Berechnung konfigurieren möchten, empfehlen wir, eine Computekonfiguration zu verwenden.

Erkennen von Abfragen für extrem große Datasets

Eine andere typische große Abfrage kann eine große Datenmenge aus großen Tabellen/Datasets scannen. Der Scanvorgang kann lange dauern und die Rechenressourcen stark beanspruchen (sogar das Lesen von Metadaten einer großen Hive-Tabelle kann viel Zeit in Anspruch nehmen). Sie können maxHivePartitions einstellen, um zu verhindern, dass zu viele Partitionen aus einer großen Hive-Tabelle abgerufen werden. Ebenso können Sie maxQueryTasks festlegen, um Abfragen für ein extrem großes Dataset einzuschränken.

spark.conf.set("spark.databricks.queryWatchdog.maxHivePartitions", 20000)
spark.conf.set("spark.databricks.queryWatchdog.maxQueryTasks", 20000)

Wann sollten Sie Query Watchdog aktivieren?

Query Watchdog sollte für Ad-hoc-Analysen aktiviert werden, bei denen SQL-Analysten und Data Scientists eine bestimmte Berechnung teilen und ein Administrator sicherstellen muss, dass Abfragen "schön" miteinander spielen.

Wann sollten Sie Query Watchdog deaktivieren?

Im Allgemeinen raten wir davon ab, Abfragen in einem ETL-Szenario voreilig abzubrechen, da es normalerweise keinen menschlichen Eingriff gibt, um den Fehler zu beheben. Es wird empfohlen, die Query Watchdog für alle außer ad-hoc-Analysen zu deaktivieren.