如何按顺序从Apache Spark发送消息到Kafka主题

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Spark Kafka Integration")
  .master("local[*]")  // 这里的master参数可以根据实际情况进行调整
  .getOrCreate()

val data = spark.read
  .format("csv")
  .option("header", "true")
  .load("path/to/input.csv")

val jsonData = data.toJSON

import java.util.Properties
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

val props = new Properties()
props.put("bootstrap.servers", "kafka-broker1:9092,kafka-broker2:9092")  // 替换为实际的Kafka集群地址
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

val producer = new KafkaProducer[String, String](props)

val topic = "my-topic"  // 替换为实际的Kafka主题名称

jsonData.foreach { json =>
  val record = new ProducerRecord[String, String](topic, json)
  producer.send(record)
}

producer.close()