无法在 Spark 结构化流中写入聚合输出

bio*_*ian 5 spark-streaming pyspark spark-structured-streaming

我刚刚开始使用 Spark 结构化流媒体,所以只是尝试一下。在汇总我的数据时;我无法将其写入 csv 文件。我尝试了以下不同的组合,但尚未实现写入操作。

我的样本数据是

colum,values
A,12
A,233
B,232
A,67
B,5
A,89
A,100
Run Code Online (Sandbox Code Playgroud)

作为流数据帧读取

userSchema = StructType([
     StructField("colum", StringType()),
    StructField("values", IntegerType())
])

line2 = spark \
.readStream \
.format('csv')\
.schema(userSchema)\
 .csv("/data/location")
Run Code Online (Sandbox Code Playgroud)

我正在做以下聚合计算

 save=line2.groupBy("colum").count()
Run Code Online (Sandbox Code Playgroud)

预期输出是

+-----+-----+
|colum|count|
+-----+-----+
|B    |2    |
|A    |5    |
|colum|1    |
+-----+-----+
Run Code Online (Sandbox Code Playgroud)

场景一:

 save.writeStream.format("csv").queryName("a").outputMode("append").option("path", "/xyz/saveloc").option("checkpointLocation", "/xyz/chkptloc").start()
Run Code Online (Sandbox Code Playgroud)

错误:当无水印的流式 DataFrame/DataSet 上存在流式聚合时,不支持追加输出模式;;

备注:由于数据中没有时间戳,因此无法添加水印。

场景2:

save.writeStream.format("csv").queryName("a").outputMode("complete").option("path", "/xyz/saveloc").option("checkpointLocation", "/xyz/chkptloc").start()
Run Code Online (Sandbox Code Playgroud)

错误::org.apache.spark.sql.AnalysisException:数据源csv不支持完整输出模式;

场景3:

save.writeStream.format("csv").queryName("a").outputMode("update").option("path", "/xyz/saveloc").option("checkpointLocation", "/xyz/chkptloc").start()
Run Code Online (Sandbox Code Playgroud)

错误:org.apache.spark.sql.AnalysisException:数据源csv不支持更新输出模式;

场景四:

save.writeStream.format("parquet").queryName("a").outputMode("update").option("path", "/xyz/saveloc").option("checkpointLocation", "/xyz/chkptloc"").start()
Run Code Online (Sandbox Code Playgroud)

错误:org.apache.spark.sql.AnalysisException:数据源镶木地板不支持更新输出模式;

场景5:

save.writeStream.format("console").queryName("a").outputMode("complete").option("path", "/xyz/saveloc").option("checkpointLocation", "/xyz/chkptloc"").start()
Run Code Online (Sandbox Code Playgroud)

注释:该位置没有生成输出。

场景六:

save.writeStream.format("memory").queryName("a").outputMode("complete").option("path", "/xyz/saveloc").option("checkpointLocation", "/xyz/chkptloc"").start()
Run Code Online (Sandbox Code Playgroud)

评论:没有生成输出。

场景7:

save.writeStream.format("memory").queryName("a").outputMode("update").option("path", "/xyz/saveloc").option("checkpointLocation", "/xyz/chkptloc"").start()
Run Code Online (Sandbox Code Playgroud)

评论:没有生成输出。

请建议我合适的配置。