Změna rozdělení stavu na vyžádání pro stavové dotazy streamování

Important

Tato funkce je ve verzi Public Preview.

Změna rozdělení stavu na vyžádání umožňuje změnit velikost počtu oddílů pro stavový dotaz strukturovaného streamování bez ztráty stavu kontrolního bodu.

Bez reparticionování stavu na vyžádání nastavíte počet shuffle oddílů při vytváření checkpointu. Pokud změníte spark.sql.shuffle.partitions, dotazy s existujícími kontrolními body novou hodnotu ignorují. Použití nového počtu oddílů vyžaduje restartování dotazu s novým kontrolním bodem.

Přerozdělení stavu na vyžádání má tyto výhody:

  • Vylaďte dotazy změnou velikosti počtu oddílů bez opětovného sestavení kontrolního bodu.
  • Škálujte dotazy nahoru nebo dolů podle změn zatížení.

Requirements

Změňte počet oddílů

Pomocí konfigurace Sparku spark.sql.streaming.stateStore.partitions a restartování dotazu změňte počet oddílů pro shuffle a stav streamování:

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()

U stavových dotazů spark.sql.streaming.stateStore.partitions má přednost před spark.sql.shuffle.partitions. Po restartování dotazu a dokončení posledního plánovaného mikrobatchu spustí dotaz operaci dělení, která redistribuuje stavová data do nového počtu oddílů. Po dokončení operace opětovného dělení se obnoví zpracování dotazu.

Monitorování stavu dělení

Po dokončení další mikrodávky zahrnují události StreamingQueryProgress i dobu trvání operace přerozdělení. V metrikách durationMs události controlBatch.REPARTITION zobrazuje hodnotu doby trvání v milisekundách. Větší velikosti stavů můžou prodloužit dobu opětovného rozdělení. Viz Monitorování dotazů strukturovaného streamování v Azure Databricks.

Příklad strukturovaného streamování

Následující příklad sníží počet oddílů shuffle u dotazu z výchozí hodnoty 200 na 100. Zastavte dotaz, nastavte počet nových oddílů a restartujte:

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()

Příklad kanálů Lakeflow

V kanálech Lakeflow nastavte spark.sql.streaming.stateStore.partitions pomocí parametru spark_conf v dekorátoru @dp.table nebo @dp.append_flow.

Nastavit oddíly pro tok:

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())

Nastavte oddíly na úrovni tabulky pro výchozí tok:

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