时间戳

首页 > 解决方案 > 使用 spark-streaming 将数据发布到 kafka 主题时重复

apache-spark - 使用 spark-streaming 将数据发布到 kafka 主题时重复

问题描述

我有 spark-streaming 应用程序，它使用来自 topic1 的数据并解析它，然后将相同的记录发布到 2 个进程中，一个是 topic2，另一个是 hive 表。在将数据发布到 kafka topic2 时，我看到重复项，但在配置单元表中看不到重复项

使用火花 2.2，卡夫卡 0.10.0

KafkaWriter.write(spark, storeSalesStreamingFinalDF, config)
writeToHIVE(spark, storeSalesStreamingFinalDF, config)


object KafkaWriter {

  def write(spark: SparkSession, df: DataFrame, config: Config)
  {
    df.select(to_json(struct("*")) as 'value)
      .write
      .format("kafka")
      .option("kafka.bootstrap.servers", config.getString("kafka.dev.bootstrap.servers"))
      .option("topic",config.getString("kafka.topic"))
      .option("kafka.compression.type",config.getString("kafka.compression.type"))
      .option("kafka.session.timeout.ms",config.getString("kafka.session.timeout.ms"))
      .option("kafka.request.timeout.ms",config.getString("kafka.request.timeout.ms"))
      .save()
  }
}

有人可以帮忙吗，

预计 kafka topic2 中没有重复项。

标签： apache-sparkapache-kafkaspark-streamingkafka-producer-apispark-streaming-kafka

解决方案

要处理重复数据，我们应该设置.option("kafka.processing.guarantee","exactly_once"）

推荐阅读