Чтение состояния Structured Streaming как таблицы через State Data Source в Spark 4

# task1.py · PySpark 4.0.4 · Spark 4.0.4
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder.appName("task1")
         .config("spark.sql.shuffle.partitions", "4")
         .getOrCreate())
spark.sparkContext.setLogLevel("WARN")

IN, CKPT = "/tmp/t1_in", "/tmp/t1_ckpt"


def add_batch(words):
    df = spark.createDataFrame([(w,) for w in words], "word string")
    df.write.mode("append").json(IN)


def run_once():
    src = spark.readStream.schema("word string").json(IN)
    q = (src.groupBy("word").count()
         .writeStream.outputMode("complete")
         .format("noop")
         .option("checkpointLocation", CKPT)
         .trigger(availableNow=True).start())
    q.awaitTermination()
    return q.lastProgress["batchId"]


def counts(batch, col_name):
    st = (spark.read.format("statestore")
          .option("batchId", batch).load(CKPT))
    return st.select(F.col("key.word").alias("word"),
                     F.col("value.count").alias(col_name))


add_batch(["a", "a", "b", "c"])
b1 = run_once()
add_batch(["a", "c", "c", "d"])
b2 = run_once()
print("N-1 =", b1, "N =", b2)
new = counts(b2, "count_N")
old = counts(b1, "count_N_minus_1")
new.join(old, "word", "full_outer").orderBy("word").show()

WARNING: Using incubator modules: jdk.incubator.vector
Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties
26/11/01 19:31:42 INFO SparkContext: Running Spark version 4.0.4
26/11/01 19:31:42 INFO SparkContext: OS info Linux, 7.0.0-1014-aws, amd64
26/11/01 19:31:42 INFO SparkContext: Java version 17.0.19
26/11/01 19:31:42 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting Spark log level to "WARN".
26/11/01 19:31:52 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.
26/11/01 19:31:56 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.
26/11/01 19:31:57 WARN HDFSBackedStateStoreProvider: The state for version 1 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:57 WARN HDFSBackedStateStoreProvider: The state for version 1 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:57 WARN HDFSBackedStateStoreProvider: The state for version 1 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:57 WARN HDFSBackedStateStoreProvider: The state for version 1 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
N-1 = 0 N = 1
26/11/01 19:31:58 WARN HDFSBackedStateStoreProvider: The state for version 1 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:58 WARN HDFSBackedStateStoreProvider: The state for version 1 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:58 WARN HDFSBackedStateStoreProvider: The state for version 1 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:58 WARN HDFSBackedStateStoreProvider: The state for version 1 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:58 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:58 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:58 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:31:58 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+----+-------+---------------+
|word|count_N|count_N_minus_1|
+----+-------+---------------+
|   a|      3|              2|
|   b|      1|              1|
|   c|      3|              1|
|   d|      1|           NULL|
+----+-------+---------------+

Разница между двумя соседними версиями видна сразу. Слово a выросло с 2 до 3, c с 1 до 3, d появилось впервые (NULL в старой версии), а b не изменилось. Здесь же batchId был 0 и 1, а в логе встречаются версии 1 и 2. Какая версия относится к какому из двух чтений, лог не различает, поэтому соответствие версий и номеров батчей прогон не показал. Зато на чистых каталогах счётчики совпали с ожидаемыми значениями.

Сделаем три запуска. Первый кладёт в состояние три ключа со временем около 12:00, второй добавляет событие на 13:00, третий добавляет событие на 13:01. Смотрим, сколько ключей остаётся после каждого запуска.

# task2.py · PySpark 4.0.4 · Spark 4.0.4
from datetime import datetime
from pyspark.sql import SparkSession

spark = (SparkSession.builder.appName("task2")
         .config("spark.sql.shuffle.partitions", "4")
         .config("spark.sql.session.timeZone", "UTC")
         .getOrCreate())
spark.sparkContext.setLogLevel("WARN")

IN, CKPT = "/tmp/t2_in", "/tmp/t2_ckpt"


def add_events(rows):
    df = spark.createDataFrame(rows, "id string, ts timestamp")
    df.write.mode("append").json(IN)


def run_once():
    src = spark.readStream.schema("id string, ts timestamp").json(IN)
    dedup = (src.withWatermark("ts", "10 minutes")
             .dropDuplicates(["id", "ts"]))
    q = (dedup.writeStream.outputMode("append")
         .format("noop")
         .option("checkpointLocation", CKPT)
         .trigger(availableNow=True).start())
    q.awaitTermination()


def show_state(title):
    print(title)
    st = spark.read.format("statestore").load(CKPT)
    st.groupBy("partition_id").count().orderBy("partition_id").show()
    st.select("key.id", "key.ts").orderBy("ts", "id").show(truncate=False)


add_events([("a", datetime(2025, 1, 1, 12, 0)),
            ("b", datetime(2025, 1, 1, 12, 0)),
            ("c", datetime(2025, 1, 1, 12, 1))])
run_once()
show_state("после запуска 1")

add_events([("d", datetime(2025, 1, 1, 13, 0))])
run_once()
show_state("после запуска 2")

add_events([("e", datetime(2025, 1, 1, 13, 1))])
run_once()
show_state("после запуска 3")

WARNING: Using incubator modules: jdk.incubator.vector
Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties
26/11/01 19:32:02 INFO SparkContext: Running Spark version 4.0.4
26/11/01 19:32:02 INFO SparkContext: OS info Linux, 7.0.0-1014-aws, amd64
26/11/01 19:32:02 INFO SparkContext: Java version 17.0.19
26/11/01 19:32:03 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting Spark log level to "WARN".
26/11/01 19:32:13 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.
после запуска 1
26/11/01 19:32:17 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:17 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:17 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:17 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+------------+-----+
|partition_id|count|
+------------+-----+
|           0|    2|
|           2|    1|
+------------+-----+

26/11/01 19:32:18 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:18 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:18 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:18 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+---+-------------------+
|id |ts                 |
+---+-------------------+
|a  |2025-01-01 12:00:00|
|b  |2025-01-01 12:00:00|
|c  |2025-01-01 12:01:00|
+---+-------------------+

26/11/01 19:32:19 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.
26/11/01 19:32:19 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:19 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:19 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:19 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
после запуска 2
26/11/01 19:32:20 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:20 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:20 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:20 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+------------+-----+
|partition_id|count|
+------------+-----+
|           2|    1|
+------------+-----+

26/11/01 19:32:21 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:21 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:21 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:21 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+---+-------------------+
… (пропущено строк: 6)
26/11/01 19:32:22 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:22 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:22 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:22 WARN HDFSBackedStateStoreProvider: The state for version 4 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
после запуска 3
26/11/01 19:32:22 WARN HDFSBackedStateStoreProvider: The state for version 6 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:22 WARN HDFSBackedStateStoreProvider: The state for version 6 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:22 WARN HDFSBackedStateStoreProvider: The state for version 6 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:22 WARN HDFSBackedStateStoreProvider: The state for version 6 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+------------+-----+
|partition_id|count|
+------------+-----+
|           0|    1|
|           2|    1|
+------------+-----+

26/11/01 19:32:23 WARN HDFSBackedStateStoreProvider: The state for version 6 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:23 WARN HDFSBackedStateStoreProvider: The state for version 6 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:23 WARN HDFSBackedStateStoreProvider: The state for version 6 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:23 WARN HDFSBackedStateStoreProvider: The state for version 6 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+---+-------------------+
|id |ts                 |
+---+-------------------+
|d  |2025-01-01 13:00:00|
|e  |2025-01-01 13:01:00|
+---+-------------------+

Прогон показывает следующее. После запуска 1 в состоянии три ключа (a, b и c). После запуска 2 число ключей сократилось с трёх до одного, и этот ключ лежит в партиции 2. Какой именно это ключ, вывод не показывает, потому что строки этого чтения в блоке пропущены. После запуска 3 в таблице два ключа, d и e. Такая картина согласуется с очисткой по watermark, ведь события около 12:00 ушли, когда появилось событие на 13:00. Но сам механизм прогон не доказывает, потому что между запусками мы состояние не читали и watermark не выводили. Версии состояния в этих трёх чтениях были 2, 4 и 6. Распределение по partition_id тут ничего не объясняет, оно лишь показывает, в каких партициях лежат ключи.

Ключ дедупликации в этом запросе включает время события (id и ts). Что произошло бы при ключе только id, мы отдельным прогоном не проверяли и утверждать не берёмся.

# task3.py · PySpark 4.0.4 · Spark 4.0.4
from datetime import datetime
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder.appName("task3")
         .config("spark.sql.shuffle.partitions", "4")
         .config("spark.sql.session.timeZone", "UTC")
         .getOrCreate())
spark.sparkContext.setLogLevel("WARN")

IN, CKPT = "/tmp/t3_in", "/tmp/t3_ckpt"

rows = [("e1", "a", datetime(2025, 1, 1, 12, 0)),
        ("e2", "a", datetime(2025, 1, 1, 12, 1)),
        ("e1", "a", datetime(2025, 1, 1, 12, 0)),
        ("e3", "b", datetime(2025, 1, 1, 12, 2))]
spark.createDataFrame(rows, "id string, word string, ts timestamp") \
    .write.mode("append").json(IN)

src = spark.readStream.schema("id string, word string, ts timestamp").json(IN)
agg = (src.withWatermark("ts", "10 minutes")
       .dropDuplicates(["id", "ts"])
       .groupBy(F.window("ts", "10 minutes"), "word").count())
q = (agg.writeStream.outputMode("append")
     .format("noop")
     .option("checkpointLocation", CKPT)
     .trigger(availableNow=True).start())
q.awaitTermination()

meta = spark.read.format("state-metadata").load(CKPT)
meta.show(truncate=False)
for row in meta.orderBy("operatorId").collect():
    op = row["operatorId"]
    print("operatorId =", op, "operatorName =", row["operatorName"])
    st = (spark.read.format("statestore")
          .option("operatorId", op).load(CKPT))
    st.printSchema()
    st.show(truncate=False)

WARNING: Using incubator modules: jdk.incubator.vector
Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties
26/11/01 19:32:27 INFO SparkContext: Running Spark version 4.0.4
26/11/01 19:32:27 INFO SparkContext: OS info Linux, 7.0.0-1014-aws, amd64
26/11/01 19:32:27 INFO SparkContext: Java version 17.0.19
26/11/01 19:32:27 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting Spark log level to "WARN".
26/11/01 19:32:37 WARN ResolveWriteToStream: spark.sql.adaptive.enabled is not supported in streaming DataFrames/Datasets and will be disabled.
+----------+--------------+--------------+-------------+----------+----------+------------------+
|operatorId|operatorName  |stateStoreName|numPartitions|minBatchId|maxBatchId|operatorProperties|
+----------+--------------+--------------+-------------+----------+----------+------------------+
|0         |stateStoreSave|default       |4            |0         |1         |NULL              |
|1         |dedupe        |default       |4            |0         |1         |NULL              |
+----------+--------------+--------------+-------------+----------+----------+------------------+

operatorId = 0 operatorName = stateStoreSave
root
 |-- key: struct (nullable = true)
 |    |-- window: struct (nullable = false)
 |    |    |-- start: timestamp (nullable = true)
 |    |    |-- end: timestamp (nullable = true)
 |    |-- word: string (nullable = true)
 |-- value: struct (nullable = true)
 |    |-- count: long (nullable = false)
 |-- partition_id: integer (nullable = true)

26/11/01 19:32:43 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:43 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:43 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:43 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+-----------------------------------------------+-----+------------+
|key                                            |value|partition_id|
+-----------------------------------------------+-----+------------+
|{{2025-01-01 12:00:00, 2025-01-01 12:10:00}, a}|{2}  |3           |
|{{2025-01-01 12:00:00, 2025-01-01 12:10:00}, b}|{1}  |3           |
+-----------------------------------------------+-----+------------+

operatorId = 1 operatorName = dedupe
root
 |-- key: struct (nullable = true)
 |    |-- id: string (nullable = true)
 |    |-- ts: timestamp (nullable = true)
 |-- value: struct (nullable = true)
 |    |-- __dummy__: void (nullable = true)
 |-- partition_id: integer (nullable = true)

26/11/01 19:32:44 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:44 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:44 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
26/11/01 19:32:44 WARN HDFSBackedStateStoreProvider: The state for version 2 doesn't exist in loadedMaps. Reading snapshot file and delta files if needed...Note that this is normal for the first batch of starting query.
+-------------------------+------+------------+
|key                      |value |partition_id|
+-------------------------+------+------------+
|{e1, 2025-01-01 12:00:00}|{NULL}|1           |
|{e3, 2025-01-01 12:02:00}|{NULL}|1           |
|{e2, 2025-01-01 12:01:00}|{NULL}|2           |
+-------------------------+------+------------+

Метаданные показывают два оператора. Оператор 0 это stateStoreSave с оконной агрегацией, ключ у него содержит окно и слово, а значение счётчик. Оператор 1 это dedupe, ключ у него пара id и ts, а значение пустое. Дубликат события e1 в состояние попал один раз, поэтому ключей три при четырёх входных строках. В агрегации счётчик слова a равен 2, потому что e1 и e2 различны, а повтор e1 отброшен дедупликацией. Здесь счётчики согласуются с входом, в отличие от основного примера. Без опции operatorId чтение в таком запросе неоднозначно, поэтому её стоит задавать явно.