如何在Scala中使用Beam使用Kafka to MySQL

libraryDependencies += "org.apache.beam" %% "beam-sdks-java-io-kafka" % "2.32.0"
libraryDependencies += "org.apache.beam" %% "beam-sdks-java-io-jdbc" % "2.32.0"

import org.apache.beam.sdk.Pipeline
import org.apache.beam.sdk.io.kafka.KafkaIO
import org.apache.beam.sdk.options.{PipelineOptions, PipelineOptionsFactory}
import org.apache.beam.sdk.transforms.{DoFn, ParDo}
import java.util.Properties

object KafkaToMySQL {
  def main(args: Array[String]): Unit = {
    // 创建 PipelineOptions 对象
    val options: PipelineOptions = PipelineOptionsFactory.create()

    // 创建 Pipeline 对象
    val pipeline: Pipeline = Pipeline.create(options)

    // Kafka 配置参数
    val kafkaProperties = new Properties()
    kafkaProperties.setProperty("bootstrap.servers", "your-kafka-server:9092")
    kafkaProperties.setProperty("group.id", "your-consumer-group")

    // 从 Kafka 读取数据
    val kafkaSource = KafkaIO
      .read[String, String]()
      .withBootstrapServers("your-kafka-server:9092")
      .withTopics(List("your-topic"))
      .withConsumerConfigUpdates(kafkaProperties)

    // 将数据写入 MySQL
    val jdbcUrl = "jdbc:mysql://your-mysql-server:3306/your-database"
    val jdbcDriver = "com.mysql.jdbc.Driver"
    val jdbcUsername = "your-username"
    val jdbcPassword = "your-password"

    kafkaSource.apply(ParDo.of(new DoFn[KV[String, String], Void] {
      @ProcessElement
      def processElement(c: DoFn[KV[String, String], Void]#ProcessContext): Unit = {
        val record = c.element()
        val key = record.getKey
        val value = record.getValue

        // 写入 MySQL
        Class.forName(jdbcDriver)
        val connection = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword)
        val statement = connection.prepareStatement("INSERT INTO your-table (key, value) VALUES (?, ?)")
        statement.setString(1, key)
        statement.setString(2, value)
        statement.executeUpdate()

        statement.close()
        connection.close()
      }
    }))

    // 运行 Pipeline
    pipeline.run().waitUntilFinish()
  }
}