apache-spark - 如何以控制台格式编写具有不同数据帧的相同流？

Question

由于我是火花结构流式传输的新手，并且在一个简单的场景中面临问题：

我正在尝试用两个不同的数据帧编写一个流。

from pyspark.sql import functions as f
df = spark \
        .readStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", "localhost:9092") \
        .option("subscribe", "topic1") \
        .option("failOnDataLoss", False)\
        .option("startingOffsets", "earliest") \
        .load()



data1 = df.filter(f.col('status') == 'true')
data2 = df.filter(f.col('status') == 'false')
data2 = data2.select(df.id,f.struct(df.col1, df.col2, df.col3).alias('value')) 
data2 = data2.groupBy("id").agg(f.collect_set('value').alias('history'))
data1 = data1.writeStream.format("console").option("truncate", "False").trigger(processingTime='15 seconds').start()
data2 = data2.writeStream.format("console").option("truncate", "False").trigger(processingTime='15 seconds').start()

spark.streams.awaitAnyTermination()

我也遇到了同样的错误：

Traceback (most recent call last):
File "/home/adarshbajpai/Downloads/spark-2.4.4-bin-hadoop2.7/python/lib/pyspark.zip/pyspark     /sql/utils.py", line 63, in deco
 File "/home/adarshbajpai/Downloads/spark-2.4.4-bin-hadoop2.7/python/lib/py4j-0.10.7-src.zip/py4j/protocol.py", line 328, in get_return_value
 py4j.protocol.Py4JJavaError: An error occurred while calling o186.start.
 : org.apache.spark.sql.AnalysisException: Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark;;
 Aggregate [customerid#93L], [customerid#93L, collect_set(hist_value#278, 0, 0) AS histleadstatus#284]
 +- Project [customerid#93L, named_struct(islaststatus, islaststatus#46, statusid, statusid#43, status, statusname#187, createdOn, statusCreatedDate#59, updatedOn, statusUpdatedDate#60) AS hist_value#278]
  +- Filter (islaststatus#46 = 0)

我认为我不应该使用水印，因为我的流媒体没有延迟和任何延迟。

请建议！提前致谢。

标签： apache-sparkpysparkapache-kafkaspark-structured-streaming