spark streaming是否能够在数据库中存储每个批次的数据？

import java.sql.{Connection, DriverManager, PreparedStatement}

// 自定义输出操作，将数据写入MySQL数据库
class MySQLSink(url: String, username: String, password: String) extends org.apache.spark.streaming.dstream.ForEachDStream[String] {

  override def foreachRDD(rdd: RDD[String]): Unit = {
    rdd.foreachPartition { partitionOfRecords =>
      val connection = DriverManager.getConnection(url, username, password)
      val statement = connection.prepareStatement("INSERT INTO my_table (data) VALUES (?)")
      partitionOfRecords.foreach { record =>
        statement.setString(1, record)
        statement.addBatch()
      }
      statement.executeBatch()
      statement.close()
      connection.close()
    }
  }
}

// 创建Spark Streaming上下文
val sparkConf = new SparkConf().setAppName("SparkStreamingExample")
val streamingContext = new StreamingContext(sparkConf, Seconds(1))

// 从数据源创建DStream
val lines = streamingContext.socketTextStream("localhost", 9999)

// 将每个批次的数据存储到MySQL数据库
lines.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    val connection = DriverManager.getConnection("jdbc:mysql://localhost/mydatabase", "username", "password")
    val statement = connection.prepareStatement("INSERT INTO my_table (data) VALUES (?)")
    partitionOfRecords.foreach { record =>
      statement.setString(1, record)
      statement.addBatch()
    }
    statement.executeBatch()
    statement.close()
    connection.close()
  }
}

// 启动StreamingContext
streamingContext.start()
streamingContext.awaitTermination()