将kafka主题中的数据读入spark dataframe

import org.apache.spark.sql.SparkSession

// 创建Spark会话
val spark = SparkSession.builder()
  .appName("KafkaSparkIntegration")
  .getOrCreate()

// 导入所需的依赖
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

// 配置Kafka参数
val kafkaParams = Map(
  "bootstrap.servers" -> "kafka-server1:9092,kafka-server2:9092",
  "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
  "group.id" -> "kafka-consumer-group",
  "auto.offset.reset" -> "latest",
  "enable.auto.commit" -> "false"
)

// 创建Kafka消费者
val kafkaConsumer = spark.readStream
  .format("kafka")
  .options(kafkaParams)
  .option("subscribe", "kafka-topic")
  .load()

// 读取Kafka数据到DataFrame
val kafkaData = kafkaConsumer.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]
  .toDF("key", "value")

// 对Kafka数据进行处理或分析
// ...

// 启动流处理
val query = kafkaData.writeStream
  .outputMode("append")
  .format("console")
  .start()

// 等待流处理完成
query.awaitTermination()