我有 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 中没有重复项。