Repartitionnement d’état à la demande pour les requêtes de streaming avec état

Important

Cette fonctionnalité est disponible en préversion publique.

Le repartitionnement d’état à la demande vous permet de redimensionner le nombre de partitions pour une requête Structured Streaming avec état sans perdre l’état du point de contrôle.

En l’absence de repartitionnement d’état à la demande, vous définissez le nombre de partitions de shuffle lors de la création du point de contrôle. Si vous changez spark.sql.shuffle.partitions, les requêtes avec des points de contrôle existants ignorent la nouvelle valeur. L’application d’un nouveau nombre de partitions vous oblige à redémarrer la requête avec un nouveau point de contrôle.

Le repartitionnement d’état à la demande présente les avantages suivants :

  • Ajustez les requêtes en redimensionnant le nombre de partitions sans regénérer le point de contrôle.
  • Ajustez les requêtes à la hausse ou à la baisse en fonction de l’évolution de la charge de travail.

Requirements

Modifier le nombre de partitions

Utilisez la configuration Spark spark.sql.streaming.stateStore.partitions et redémarrez la requête pour modifier le nombre de partitions de shuffle et d’état de streaming :

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

Pour les requêtes avec état, spark.sql.streaming.stateStore.partitions est prioritaire sur spark.sql.shuffle.partitions. Une fois la requête redémarrée et la dernière microbatch planifiée terminée, la requête exécute une opération de repartition pour redistribuer les données d’état dans le nouveau nombre de partitions. Une fois l’opération de repartition terminée, la requête reprend le traitement.

État de repartition du moniteur

Une fois le microbatch suivant terminé, les événements StreamingQueryProgress incluent la durée de l’opération de repartition. Dans les métriques d’un durationMs événement, controlBatch.REPARTITION affiche la valeur de durée en millisecondes. Des tailles d’état plus importantes peuvent augmenter le temps nécessaire au repartitionnement. Consultez la surveillance des requêtes de streaming structuré sur Azure Databricks.

Exemple de streaming structuré

L’exemple suivant réduit le nombre de partitions de shuffle d’une requête de 200, la valeur par défaut, à 100. Arrêtez la requête, définissez le nouveau nombre de partitions et redémarrez :

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

Exemple de pipelines Lakeflow

Dans les pipelines Lakeflow, définissez spark.sql.streaming.stateStore.partitions à l’aide du paramètre spark_conf sur le décorateur @dp.table ou @dp.append_flow.

Définir des partitions pour un flux :

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

Définissez des partitions au niveau de la table pour le flux par défaut :

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