从Kafka主题读取处理数据并使用scala和spark写回Kafka主题

嗨，我正在阅读 kafka 主题，我想处理从 kafka 收到的数据，例如标记化、过滤掉不必要的数据、删除停用词，最后我想写回另一个 Kafka 主题

// read from kafka
val readStream = existingSparkSession
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", hostAddress)
      .option("subscribe", "my.raw") // Always read from offset 0, for dev/testing purpose
      .load()

val df = readStream.selectExpr("CAST(value AS STRING)" )
df.show(false)
val df_json = df.select(from_json(col("value"), mySchema.defineSchema()).alias("parsed_value"))
val df_text = df_json.withColumn("text", col("parsed_value.payload.Text"))

// perform some data processing actions such as tokenization etc and return cleanedDataframe as the final result

// write back to kafka
val writeStream = cleanedDataframe
      .writeStream
      .outputMode("append")
      .format("kafka")
      .option("kafka.bootstrap.servers", hostAddress)
      .option("topic", "writing.val")
      .start()
    writeStream.awaitTermination()

然后我收到以下错误

线程“main” org.apache.spark.sql.AnalysisException 中的异常：必须使用 writeStream.start(); 执行具有流源的查询；

然后我编辑了我的代码，如下所示从 kafka 读取并写入控制台

// read from kafka
val readStream = existingSparkSession
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", hostAddress)
      .option("subscribe", "my.raw") // Always read from offset 0, for dev/testing purpose
      .load()

// write to console
val df = readStream.selectExpr("CAST(value AS STRING)" )
val query = df.writeStream
      .outputMode("append")
      .format("console")
      .start().awaitTermination();

// then perform the data processing part as mentioned in the first half

使用第二种方法，在控制台中连续显示数据，但它从未运行过数据处理部分。我可以知道如何从 kafka 主题中读取数据，然后对接收到的数据执行一些操作（标记化、删除停用词），最后写回新的 kafka 主题吗？

编辑

堆栈跟踪在错误期间指向上述代码中的 df.show(false)

回答

// read from kafka
val readStream = existingSparkSession
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", hostAddress)
      .option("subscribe", "my.raw") // Always read from offset 0, for dev/testing purpose
      .load()

// write to console
val df = readStream.selectExpr("CAST(value AS STRING)" )
df.writeStream
      .outputMode("append")
      .format("console")
      .start()

val df_json = df.select(from_json(col("value"), mySchema.defineSchema()).alias("parsed_value"))
val df_text = df_json.withColumn("text", col("parsed_value.payload.Text"))

// perform some data processing actions such as tokenization etc and return cleanedDataframe as the final result

// write back to kafka
// the columns `key` and `value` of the DataFrame `cleanedDataframe` will be used for producing the message into the Kafka topic.
val writeStreamKafka = cleanedDataframe
      .writeStream
      .outputMode("append")
      .format("kafka")
      .option("kafka.bootstrap.servers", hostAddress)
      .option("topic", "writing.val")
      .start()

existingSparkSession.awaitAnyTermination()