从Kafka主题读取数据并使用scala和火花写回Kafka主题

Question

嗨，我正在阅读kafka主题，我想处理从kafka接收到的数据，例如tockenize，过滤掉不必要的数据，删除停用词，最后我想写回另一个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.Analysis Exception：具有流源的查询必须使用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主题？

编辑

Stack Trace在错误期间指向上面代码中的df.show（false）

Answer 1

当前实现中有两个常见问题：

1. 在流上下文中应用show
2. 等待终止后的代码将不会被执行

到1。

show方法是对数据帧执行的操作（与转换相对）。当您处理流式数据帧时，这将导致错误，因为流式查询需要使用开始（正如expetion文本所告诉您的那样）。

至2。

方法waittermination是一种阻塞方法，这意味着后续代码不会在每个微批中执行。

总体解决方案

如果您想对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)" )
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()

请注意现有SparkSession。waitanytermination（）位于代码的最末尾，而不直接在启动后使用waitanytermination。此外，请记住，数据框的键和值列将用于将消息生成到Kafka主题中。但是，不需要列键，另请参见此处

此外，如果您使用检查点（推荐），那么您需要设置两个不同的位置：一个用于控制台流，另一个用于kafka输出流。重要的是要记住这些流查询独立运行。