写入 Kafka 主题不起作用但没有错误答案

【问题标题】：Writing to Kafka topic not working but no error写入 Kafka 主题不起作用但没有错误
【发布时间】：2021-12-20 17:14:52
【问题描述】：

我使用文件源读取流数据

parquet_sdf = spark.readStream.schema(schema).parquet(path)

然后我对df做一些变换

sdf = parquet_sdf. \
        withColumn('time', current_timestamp()). \
        withWatermark('time', '5 seconds'). \
        withColumn('country', substring('monitoringSiteIdentifier', 1, 2)). \
        where(col('resultObservationStatus') == 'A').\
        groupBy('country', 'resultObservationStatus', 'time').count()

并尝试将其存储在 Kafka 中，但它不起作用。

sdf \
        .selectExpr("CAST(country AS STRING) AS key", "to_json(struct(*)) AS value") \
        .writeStream \
        .format("kafka") \
        .option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS) \
        .option("topic", TOPIC) \
        .option("checkpointLocation", "/tmp/demo") \
        .trigger(processingTime='1 seconds') \
        .start()

我有一个将文件移动到 parquet 目录路径的脚本，然后我运行此代码。我没有收到任何错误。但是看不到关于 Kafka 主题的任何消息。当我尝试下面的控制台格式时，它可以工作，我可以在控制台上看到消息

sdf.writeStream \
         .trigger(processingTime='1 seconds') \
         .outputMode("update") \
         .option("truncate", "false") \
         .format("console") \
         .start().awaitTermination()

我不确定为什么它在 Kafka 中不起作用。

【问题讨论】：

如果你已经写入控制台，那么文件已经被“处理”了，所以当你重新启动代码时，Spark 不会尝试写入 Kafka。文件源也只适用于原子移动文件
@OneCricketeer：我不会同时做这两件事。同样在运行脚本之前，我清除所有 checkpointLocation 并确保它为空，然后在脚本运行时将文件复制到 readStream 目录。

标签： apache-spark pyspark apache-kafka spark-structured-streaming

【解决方案1】：

我还需要在writeStream 方法的末尾添加.awaitTermination()。

sdf.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
        .writeStream \
        .trigger(processingTime='1 seconds') \
        .format("kafka") \
        .option("kafka.bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS) \
        .option("topic", TOPIC) \
        .option("checkpointLocation", "/tmp/demo") \
        .start() \
        .awaitTermination()

然后在运行脚本之前，我清除所有checkpointLocation 并确保它为空，然后将文件复制到脚本运行时readStream 正在读取的目录中，然后在kafka-console-consumer 中等待窗口，它开始出现。

【讨论】：