为什么使用更新输出模式的流式查询会打印出所有行？

Question

我的目录中有三个文本文件：

一个.txt

A B
C D
A E
F

b.txt

A B
C D
A E

文件

A B
C D
A E
G

我使用以下流查询：

val schema = new StructType().add("value", "string")
val lines = spark
  .readStream
  .schema(schema)
  .option("maxFilesPerTrigger", 1)
  .text(...)
  .as[String]

val wordCounts = lines.flatMap(_.split("\\s+")).groupBy("value").count()

val query = wordCounts.writeStream
  .queryName("t")
  .outputMode("update") // <-- output mode: update
  .format("memory")
  .start()

while (true) {
  spark.sql("select * from t").show(truncate = false)
  println(new Date())
  Thread.sleep(1000)
}

查询总是输出以下结果：

+-----+-----+
|value|count|
+-----+-----+
|A    |2    |
|B    |1    |
|C    |1    |
|D    |1    |
|E    |1    |
|A    |4    |
|B    |2    |
|C    |2    |
|D    |2    |
|E    |2    |
|G    |1    |
|A    |6    |
|B    |3    |
|C    |3    |
|D    |3    |
|E    |3    |
|F    |1    |
+-----+-----+

看起来每个文件的结果都附加到输出结果（如在Append输出模式中），我不确定我是否理解update模式的含义。update输出模式如何工作？

Answer 1

在Append模式下，只有自上次触发以来添加到结果表中的新行才会被输出到接收器。仅那些添加到结果表中的行永远不会更改的查询才支持此功能。因此，该模式保证每行仅输出一次。

更新中模式下，只有结果表中自上次触发以来更新的行才会输出到接收器。

为了更好地理解这些模式，我改变了控制台的输出格式并修改了数据，在更新模式下执行，结果如下：

a.txt
A B
C D
A E
F X
Y Z

b.txt
A B
C D
A E

c.txt
A B
C D
A E
G


scala> val query = wordCounts.writeStream.queryName("t").outputMode("update").format("console").start()
query: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@1985f8e3

scala> -------------------------------------------
Batch: 0
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
|    F|    1|
|    E|    1|
|    B|    1|
|    Y|    1|
|    D|    1|
|    C|    1|
|    Z|    1|
|    A|    2|
|    X|    1|
+-----+-----+

-------------------------------------------
Batch: 1
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
|    E|    2|
|    B|    2|
|    D|    2|
|    C|    2|
|    A|    4|
+-----+-----+

-------------------------------------------
Batch: 2
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
|    E|    3|
|    B|    3|
|    D|    3|
|    C|    3|
|    A|    6|
|    G|    1|
+-----+-----+

您可以看到，对于每个批次，只有自上次触发以来更新的行才会显示在控制台中。（例如：X、Y、Z 计数不会显示在批次 1 和 2 中，因为它们未更新）。

在您的情况下，当您将数据写入内存时。由于您没有逐出每个批次的内存，因此在查询时也会检索上一个批次的数据。希望现在模式已经清楚了。