Spark 结构化流恰好一次 - 未实现 - 重复事件答案

【问题标题】：Spark structured streaming exactly once - Not achieved - Duplicated eventsSpark 结构化流恰好一次 - 未实现 - 重复事件
【发布时间】：2019-08-12 10:20:04
【问题描述】：

我正在使用 Spark Structured Streaming 来使用来自 Kafka 的事件并将它们上传到 S3。

检查点在 S3 上提交：

DataFrameWriter<Row> writer = input.writeStream()
           .format("orc")
           .trigger(ProcessingTime(config.getProcessingTime()))
           .outputMode(OutputMode.Append())
           .option("truncate", false)           
           .option("checkpointLocation", "s3://bucket1")
           .option("compression", "zlib")
           .option("path", "s3://bucket2");

偏移量通过StreamingQueryListener 提交给Kafka：

  kafkaConsumer.commitSync(topicPartitionMap);

一旦应用程序启动，它就会从 Kafka 检索偏移图并启动流：

 reader = sparkSession
            .readStream()
            .format("kafka")
            .option("kafka.bootstrap.servers", config.getKafkaBootStrapServers())
            .option("subscribe", "topic1")
            .option("max.poll.records", 1000)
            .option("failOnDataLoss", false)
            .option("startingOffsets", topicPartitionMap)

我将 topic/partition/offset 与 ORC 文件中的数据一起存储。

数据包含事件的多个重复项，确切地说是topic/partition/offset。

应该如何配置流以实现一次处理？

【问题讨论】：

为什么要编写 Spark 代码而不是使用其他公司在生产中使用的现有服务？ github.com/pinterest/secor
多种原因 1.你需要管理schema才能用Secor写ORC，2.我们有自己的业务逻辑3.我们已经在使用Spark集群了。
你也可以查看这个项目：github.com/YotpoLtd/metorikku
您是否愿意考虑使用 Apache Spark 以外的系统？
我测试了Secor和Kafka connect，这两个不适合我

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

【解决方案1】：

发现这些参数应该设置为true spark.streaming.driver.writeAheadLog.closeFileAfterWrite 和 spark.streaming.receiver.writeAheadLog.closeFileAfterWrite

当您想将 S3 用于元数据 WAL 时，将此设置为“true”

https://spark.apache.org/docs/latest/configuration.html

【讨论】：