以下是一个简单的使用 Spark Streaming 读取 Kafka 消息、统计数据后插入到 MySQL 中的 Scala 代码示例：

import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.kafka.KafkaUtils import java.sql.DriverManager  object KafkaToMysql {    def main(args: Array[String]): Unit = {     // 创建 SparkConf     val conf = new SparkConf().setAppName("KafkaToMysql")     // 创建 StreamingContext     val ssc = new StreamingContext(conf, Seconds(5))      // 设置 Kafka 相关参数     val kafkaParams = Map[String, String](       "bootstrap.servers" -> "your_kafka_broker:9092",       "group.id" -> "your_group_id"     )      // 定义要读取的 Kafka 主题     val topics = Array("your_topic")      // 使用 KafkaUtils 创建 DStream     val kafkaStream = KafkaUtils.createDirectStream[String, String](       ssc,       LocationStrategies.PreferConsistent,       ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)     )      // 解析和统计数据     val data = kafkaStream.map(_.value()).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)      // 定义数据库连接相关信息     val url = "jdbc:mysql://your_host:your_port/your_database"     val username = "your_username"     val password = "your_password"      // 将统计结果插入到 MySQL 中     data.foreachRDD(rdd => {       rdd.foreachPartition(partition => {         Class.forName("com.mysql.jdbc.Driver")         val connection = DriverManager.getConnection(url, username, password)         val statement = connection.createStatement()         partition.foreach { case (word, count) =>           val sql = s"INSERT INTO your_table (word, count) VALUES ('$word', $count)"           statement.executeUpdate(sql)         }         connection.close()       })     })      ssc.start()     ssc.awaitTermination()   } }