将spark结构化流输出写入kafka主题

gtlvzcf8  于 2021-06-08  发布在  Kafka
关注(0)|答案(1)|浏览(306)

我有一个简单的结构化流媒体应用程序,它只从一个kafka主题读取数据,然后写入另一个主题。

SparkConf conf = new SparkConf()
        .setMaster("local[*]")
        .setAppName("test");

SparkSession spark = SparkSession
        .builder()
        .config(conf)
        .getOrCreate();

Dataset<Row> dataset = spark
        .readStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("subscribe", "start")
        .load();

StreamingQuery query = dataset
        .writeStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("checkpointLocation", "checkpoint")
        .option("topic", "end")
        .start();

query.awaitTermination(20000);

主题上有两条消息要处理 start . 这段代码毫无例外地运行,但是没有任何消息会出现在主题上 end . 这个例子有什么问题?

6ioyuze2

6ioyuze21#

问题是消息已经在流中,并且起始偏移量没有设置为“最早”。

Dataset<Row> dataset = spark
        .readStream()
        .format("kafka")
        .option("kafka.bootstrap.servers", "localhost:9092")
        .option("subscribe", start.getTopicName())
        .option("startingOffsets", "earliest")
        .load();

相关问题