如何解决Spark结构化流文件接收器创建空的JSON文件
我正在从Kafka主题中读取数据并能够处理数据。
当我尝试以.json格式存储文件时，HDFS包含空的.json文件。
“”“
KAFKA_CONFLUENT_TOPIC_REPLICATION_FACTOR这是HDFS输出：
query = KPI_Final_DF \
.writeStream \
.outputMode("Append") \
.format("json") \
.option("truncate","false") \
.option("path","output_3") \
.option("checkpointLocation","output_json") \
.trigger(processingTime="1 minute") \
.start()
# query termination command
query.awaitTermination()
"""
Below is the console output:
-------------------------------------------
Batch: 27
-------------------------------------------
+------------------------------------------+--------------+------------------+---+-------------------+
|window |country |Total_Volume_Sale |OPM|Rate_Return |
+------------------------------------------+--------------+------------------+---+-------------------+
|[2020-11-05 16:30:00,2020-11-05 16:31:00]|United Kingdom|37.010000705718994|2 |0.0 |
|[2020-11-05 16:29:00,2020-11-05 16:30:00]|United Kingdom|613.1199990212917 |11 |0.15384615384615385|
+------------------------------------------+--------------+------------------+---+-------------------+
-------------------------------------------
Batch: 28
-------------------------------------------
+------------------------------------------+--------------+-----------------+---+-----------+
|window |country |Total_Volume_Sale|OPM|Rate_Return|
+------------------------------------------+--------------+-----------------+---+-----------+
|[2020-11-05 16:30:00,2020-11-05 16:31:00]|United Kingdom|66.70999991893768|3 |0.0 |
+------------------------------------------+--------------+-----------------+---+-----------+
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。
