作者：mobiledu2402851377 | 来源：互联网 | 2023-02-01 20:36
我使用Spark 2.2.0-rc1.
我有一个卡夫卡topic这我查询运行水印聚集,有1 minute水印,发出来console与append输出模式.
import org.apache.spark.sql.types._
val schema = StructType(StructField("time", TimestampType) :: Nil)
val q = spark.
readStream.
format("kafka").
option("kafka.bootstrap.servers", "localhost:9092").
option("startingOffsets", "earliest").
option("subscribe", "topic").
load.
select(from_json(col("value").cast("string"), schema).as("value"))
select("value.*").
withWatermark("time", "1 minute").
groupBy("time").
count.
writeStream.
outputMode("append").
format("console").
start
我在Kafka推送以下数据topic:
{"time":"2017-06-07 10:01:00.000"}
{"time":"2017-06-07 10:02:00.000"}
{"time":"2017-06-07 10:03:00.000"}
{"time":"2017-06-07 10:04:00.000"}
{"time":"2017-06-07 10:05:00.000"}
我得到以下输出:
scala> -------------------------------------------
Batch: 0
-------------------------------------------
+----+-----+
|time|count|
+----+-----+
+----+-----+
-------------------------------------------
Batch: 1
-------------------------------------------
+----+-----+
|time|count|
+----+-----+
+----+-----+
-------------------------------------------
Batch: 2
-------------------------------------------
+----+-----+
|time|count|
+----+-----+
+----+-----+
-------------------------------------------
Batch: 3
-------------------------------------------
+----+-----+
|time|count|
+----+-----+
+----+-----+
-------------------------------------------
Batch: 4
-------------------------------------------
+----+-----+
|time|count|
+----+-----+
+----+-----+
这是预期的行为吗？
1> zsxwing..：
向Kafka推送更多数据应该会触发Spark输出内容.目前的行为完全是因为内部实施.
当您推送一些数据时,StreamingQuery将生成一个要运行的批处理.当该批次完成时,它将记住该批次中的最大事件时间.然后在下一批中,因为您正在使用append模式,StreamingQuery将使用最大事件时间和水印来从StateStore中逐出旧值并输出它.因此,您需要确保生成至少两个批次才能查看输出.
京公网安备 11010802041100号