Bedarfsgesteuerte Zustandsneupartitionierung für zustandsbehaftete Streamingabfragen

Von Bedeutung

Dieses Feature befindet sich in der Public Preview.

Die bedarfsgesteuerte Zustandsneupartitionierung ermöglicht es Ihnen, die Anzahl der Partitionen für eine zustandsbehaftete strukturierte Streamingabfrage zu ändern, ohne den Prüfpunktzustand zu verlieren.

Ohne bedarfsgesteuerte Zustandsneupartitionierung legen Sie die Anzahl der Shuffle-Partitionen während der Prüfpunkterstellung fest. Wenn Sie spark.sql.shuffle.partitions ändern, ignorieren Abfragen mit vorhandenen Prüfpunkten den neuen Wert. Wenn Sie eine neue Partitionsanzahl anwenden, müssen Sie die Abfrage mit einem neuen Prüfpunkt neu starten.

Die bedarfsgesteuerte Zustandsneupartitionierung bietet die folgenden Vorteile:

  • Optimieren Sie Abfragen, indem Sie die Anzahl der Partitionen ändern, ohne den Prüfpunkt neu zu erstellen.
  • Skalieren Sie Abfragen nach oben oder unten, um Arbeitsauslastungsänderungen abzugleichen.

Requirements

Ändern der Anzahl von Partitionen

Verwenden Sie die Spark-Konfiguration spark.sql.streaming.stateStore.partitions , und starten Sie die Abfrage neu, um die Anzahl der Partitionen des Shuffle- und Streamingstatus zu ändern:

Python

query.stop()
spark.conf.set("spark.sql.streaming.stateStore.partitions", "<numPartitions>")
query = df.writeStream.start()

Scala

query.stop()
spark.conf.set("spark.sql.streaming.stateStore.partitions", "<numPartitions>")
val query = df.writeStream.start()

Für zustandsbehaftete Abfragen spark.sql.streaming.stateStore.partitions hat Vorrang vor spark.sql.shuffle.partitions. Nachdem die Abfrage neu gestartet und die letzte geplante Microbatch abgeschlossen wurde, führt die Abfrage einen Neupartitionsvorgang aus, um Zustandsdaten in die neue Anzahl von Partitionen neu zu verteilen. Nach Abschluss des Repartitionsvorgangs setzt die Abfrage die Verarbeitung fort.

Status der Neupartitionierung überwachen

Nach Abschluss des nächsten Microbatches enthalten StreamingQueryProgress-Ereignisse die Dauer des Neupartitionierungsvorgangs. In den Metriken durationMs eines Ereignisses controlBatch.REPARTITION wird der Dauerwert in Millisekunden angezeigt. Eine größere Größe des Zustands kann die für die Repartitionierung benötigte Zeit verlängern. Siehe Überwachung strukturierter Streaming-Abfragen auf Azure Databricks.

Beispiel für strukturiertes Streaming

Im folgenden Beispiel wird eine Abfrage von 200, der Standard, auf 100 Shuffle-Partitionen skaliert. Beenden Sie die Abfrage, legen Sie die neue Partitionsanzahl fest, und starten Sie folgendes neu:

Python

# Start the query with the default partition count (200)
query = (df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window("event_time", "5 minutes"),
    "id")
  .count()
  .writeStream
  .format("delta")
  .option("checkpointLocation", "/checkpoint/path")
  .outputMode("append")
  .start()
)

# Stop the query and scale down to 100 partitions
query.stop()

spark.conf.set("spark.sql.streaming.stateStore.partitions", "100")

# Restart the query with the same options
query = (df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window("event_time", "5 minutes"),
    "id")
  .count()
  .writeStream
  .format("delta")
  .option("checkpointLocation", "/checkpoint/path")
  .outputMode("append")
  .start()
)

Scala

// Start the query with the default partition count (200)
val query = df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window($"event_time", "5 minutes"),
    $"id")
  .count()
  .writeStream
  .format("delta")
  .option("checkpointLocation", "/checkpoint/path")
  .outputMode("append")
  .start()

// Stop the query and scale down to 100 partitions
query.stop()

spark.conf.set("spark.sql.streaming.stateStore.partitions", "100")

// Restart the query with the same options
val query2 = df
  .withWatermark("event_time", "10 minutes")
  .groupBy(
    window($"event_time", "5 minutes"),
    $"id")
  .count()
  .writeStream
  .format("delta")
  .option("checkpointLocation", "/checkpoint/path")
  .outputMode("append")
  .start()

Beispiel für Lakeflow-Pipelines

Legen Sie in Lakeflow-Pipelines spark.sql.streaming.stateStore.partitions mithilfe des Parameters spark_conf im Decorator @dp.table oder @dp.append_flow fest.

Partitionen für einen Flow festlegen:

from pyspark import pipelines as dp
from pyspark.sql import functions as F

source_path = "/databricks-datasets/iot-stream/data-device/"

dp.create_streaming_table("target_table")

@dp.append_flow(
  target="target_table",
  name="my_flow_1",
  spark_conf={"spark.sql.streaming.stateStore.partitions": "100"}
)
def my_flow_1():
  return (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .load(source_path)
    .withColumn("timestamp", F.to_timestamp("timestamp"))
    .withWatermark("timestamp", "10 minutes")
    .groupBy(F.window("timestamp", "5 minutes"), "id")
    .count())

Legen Sie Partitionen auf Tabellenebene für den Standardfluss fest:

from pyspark import pipelines as dp
from pyspark.sql import functions as F

source_path = "/databricks-datasets/iot-stream/data-device/"

@dp.table(
  name="table_1",
  spark_conf={"spark.sql.streaming.stateStore.partitions": "100"}
)
def table_1():
  return (spark.readStream.format("cloudFiles")
    .option("cloudFiles.format", "json")
    .load(source_path)
    .withColumn("timestamp", F.to_timestamp("timestamp"))
    .withWatermark("timestamp", "10 minutes")
    .groupBy(F.window("timestamp", "5 minutes"), "id")
    .count())
  • Last updated on