用离线spark写, 不用批处理写

如果你想在离线的Spark作业中将数据写入Kafka，你可以使用`foreachPartition`函数来实现。这将允许你对数据分区进行迭代，并在每个分区上创建一个Kafka生产者来写入数据。

下面是一个使用离线Spark批处理来将数据写入Kafka的示例代码：

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.spark.sql.{SparkSession, DataFrame}

object KafkaWriter {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("KafkaWriter")
      .master("local[*]")  // 替换为你的Spark Master URL
      .getOrCreate()

    // 读取表数据
    val tableDF = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/database")  // 替换为你的数据库连接URL
      .option("dbtable", "table_name")  // 替换为你要读取的表名
      .option("user", "username")  // 替换为你的数据库用户名
      .option("password", "password")  // 替换为你的数据库密码
      .load()

    // 写入Kafka
    writeToKafka(tableDF, "your_kafka_topic")  // 替换为你的Kafka主题名

    spark.stop()
  }

  def writeToKafka(df: DataFrame, topic: String): Unit = {
    df.foreachPartition { partition =>
      val kafkaProps = new java.util.Properties()
      kafkaProps.put("bootstrap.servers", "localhost:9092")  // 替换为你的Kafka broker地址
      kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
      kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

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

      partition.foreach { row =>
        val key = row.getAs[String]("key")
        val value = row.getAs[String]("value")
        
        val record = new ProducerRecord[String, String](topic, key, value)
        producer.send(record)
      }

      producer.close()
    }
  }
}

在上述代码中，你需要替换以下内容：
- Spark的Master URL (`master`)，例如 `local[*]`。
- 数据库连接URL (`url`)，例如 `jdbc:mysql://localhost:3306/database`，其中 `localhost:3306` 是你的数据库主机和端口，`database` 是你的数据库名。
- 要读取的表名 (`dbtable`)。
- 数据库用户名 (`user`) 和密码 (`password`)。
- Kafka broker地址 (`bootstrap.servers`)，例如 `localhost:9092`。
- 要写入的Kafka主题名 (`topic`)。

请确保你已经正确配置了Kafka和表的连接信息，以及相关依赖库。运行代码后，它将使用离线Spark作业读取表数据，并将数据写入到指定的Kafka主题中。