sparkstreaming如何将数据写入mysql和 每行代码解释

将数据写入MySQL是常见的Spark Streaming数据处理过程之一。下面是一个简单的示例，演示如何将Spark Streaming处理后的数据写入MySQL数据库。

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

// 定义一个函数，用于将数据写入MySQL
def writeToMySQL(iter: Iterator[(String, Int)]): Unit = {
  var conn: Connection = null
  var ps: PreparedStatement = null

  val sql = "insert into wordcount(word, count) values (?, ?)"
  try {
    // 获取MySQL连接
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "password")
    iter.foreach(data => {
      // 每个分区创建PreparedStatement对象
      ps = conn.prepareStatement(sql)
      ps.setString(1, data._1)
      ps.setInt(2, data._2)
      ps.executeUpdate()
    })
  } catch {
    case e: Exception => e.printStackTrace()
  } finally {
    // 关闭连接和PreparedStatement对象
    if (ps != null) {
      ps.close()
    }
    if (conn != null) {
      conn.close()
    }
  }
}

// 创建StreamingContext对象
val ssc = new StreamingContext(sparkConf, Seconds(2))

// 创建DStream对象，从文本文件中读取数据
val lines = ssc.textFileStream("hdfs://localhost:9000/data")

// 对DStream进行处理
val wordCounts = lines.flatMap(_.split(" ")).map(x => (x, 1)).reduceByKey(_ + _)

// 将结果写入MySQL
wordCounts.foreachRDD(rdd => {
  rdd.foreachPartition(iter => {
    // 每个分区的数据写入MySQL
    writeToMySQL(iter)
  })
})

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

代码解释：

1. `writeToMySQL` 函数用于将数据写入MySQL。首先获取MySQL连接，然后遍历每个分区的数据，创建 `PreparedStatement` 对象并执行插入操作，最后关闭连接和 `PreparedStatement` 对象。

2. 创建 `StreamingContext` 对象，指定批处理时间间隔为 2 秒。

3. 创建 `DStream` 对象，从文本文件中读取数据。

4. 对 `DStream` 进行处理，使用 `flatMap` 和 `map` 函数将每行数据分割成单词并计数。

5. 将结果写入MySQL，使用 `foreachRDD` 函数遍历每个RDD，对每个分区的数据调用 `writeToMySQL` 函数进行写入。

6. 启动 `StreamingContext` 并等待处理完成。

希望这个示例能够帮助您了解如何将数据写入MySQL，以及Spark Streaming的基本用法。