SparkStream写入数据到mysql

客官°小女子只卖身不卖艺 2023-06-15 11:18 17阅读 0赞

## SparkStream写入数据到mysql ##

在[https://blog.csdn.net/qq\_35885488/article/details/103185724][https_blog.csdn.net_qq_35885488_article_details_103185724]基础上进行修改

建表语句

create table wordcount(
    
    word varchar(50) default null
    
    wordcount int (10) default null
    
    )；

### 使用foreachRDD的设计模式 ###

[看一段官网的介绍][Link 1]

`dstream.foreachRDD`是一个强大的语句，可以将数据发送到外部系统。但是，重要的是要了解如何正确有效地使用此语句。应避免的一些常见错误如下。

通常，将数据写入外部系统需要创建一个连接对象（例如，到远程服务器的TCP连接），并使用该对象将数据发送到远程系统。为此，开发人员可能会无意间尝试在Spark驱动程序中创建连接对象，然后尝试在Spark worker中使用该对象以将记录保存在RDD中。

例如（在Scala中）

//将结果写到mysql
        result.foreachRDD { rdd =>
          val connection = createConn()  // executed at the driver
          rdd.foreach { record =>
    
           val sql = "insert into wordcount(word,wordcount) values ('"+ record._1 +"',"+record._2+")" // executed at the worker
          connection.createStatement().execute(sql)
    
          }
        }
    
      /**
        * 获取mysql连接
        *连接池实现
        */
    def createConn() = {
      Class.forName("com.mysql.jdbc.Driver")
      DriverManager.getConnection("jdbc:mysql://localhost:3306/test","root","admin")
    }

这是不正确的，因为这要求将连接对象序列化并从驱动程序发送给工作程序。这样的连接对象很少能在机器之间转移。此错误可能表现为序列化错误（连接对象不可序列化），初始化错误（连接对象需要在工作程序中初始化）等。正确的解决方案是在工作程序中创建连接对象。

报错信息

org.apache.spark.SparkException: Task not serializable
    	at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298)
    	at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288)
    	at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108)
    	at org.apache.spark.SparkContext.clean(SparkContext.scala:2287)
    	at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:917)
    	at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:916)
    	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    	at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
    	at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
    	at org.apache.spark.rdd.RDD.foreach(RDD.scala:916)
    	at com.zgw.spark.ForEachRddApp$$anonfun$main$1.apply(ForEachRddApp.scala:35)
    	at com.zgw.spark.ForEachRddApp$$anonfun$main$1.apply(ForEachRddApp.scala:33)
    	at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:628)
    	at org.apache.spark.streaming.dstream.DStream$$anonfun$foreachRDD$1$$anonfun$apply$mcV$sp$3.apply(DStream.scala:628)
    	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ForEachDStream.scala:51)
    	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
    	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(ForEachDStream.scala:51)
    	at org.apache.spark.streaming.dstream.DStream.createRDDWithLocalProperties(DStream.scala:416)
    	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:50)
    
    Caused by: java.io.NotSerializableException: com.mysql.jdbc.SingleByteCharsetConverter
    Serialization stack:

正确的做法应该是：

result.foreachRDD(rdd =>{ 
          rdd.foreachPartition(partitionOnRecords =>{ 
            val connection = createConn()  // executed at the driver
            partitionOnRecords.foreach(record=>{ 
              val sql = "insert into wordcount(word,wordcount) values ('"+ record._1 +"',"+record._2+")" // executed at the worker
              connection.createStatement().execute(sql)
            })
            connection.close()
          })
        })

完整代码

package com.zgw.spark.streaming
    
    import java.sql.DriverManager
    
    import org.apache.spark.SparkConf
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    import org.apache.spark.streaming.dstream.DStream
    
    /** * Created by Zhaogw&Lss on 2019/11/22. */
    object ForEachRdd1 { 
      def main(args: Array[String]): Unit = { 
    
        val sparkConf = new SparkConf().setAppName("StatefulWordCount").setMaster("local[2]").set("spark.testing.memory", "2147480000")
        val ssc = new StreamingContext(sparkConf, Seconds(5))
    
        // 如果使用了stateful的算子，必须要设置checkpoint
        // 在生产环境中，建议大家把checkpoint设置到HDFS的某个文件夹中
        ssc.checkpoint(".")
    
        val lines = ssc.socketTextStream("hadoop000", 9999)
    
        val result: DStream[(String, Int)] = lines.flatMap(_.split(" ")).map((_,1))
        val state = result.updateStateByKey[Int](updateFunction _)
    
        state.print()     //输出到控制台
    
        result.foreachRDD(rdd =>{ 
          rdd.foreachPartition(partitionOnRecords =>{ 
            val connection = createConn()  // executed at the driver
            partitionOnRecords.foreach(record=>{ 
              val sql = "insert into wordcount(word,wordcount) values ('"+ record._1 +"',"+record._2+")" // executed at the worker
              connection.createStatement().execute(sql)
            })
            connection.close()
          })
        })
        ssc.start()
        ssc.awaitTermination()
      }
      /** * 获取mysql连接 *连接池实现 */
      def createConn() = { 
        Class.forName("com.mysql.jdbc.Driver")
        DriverManager.getConnection("jdbc:mysql://localhost:3306/test","root","admin")
      }
    
      /** * 把当前的数据去更新已有的或者是老的数据 * @param currentValues 当前的 * @param preValues 老的 * @return */
      def updateFunction(currentValues: Seq[Int], preValues: Option[Int]): Option[Int] = { 
        val current = currentValues.sum
        val pre = preValues.getOrElse(0)
    
        Some(current + pre)
      }
    }

托管于[https://github.com/daizikaikou/learningSpark][https_github.com_daizikaikou_learningSpark]

[https_blog.csdn.net_qq_35885488_article_details_103185724]: https://blog.csdn.net/qq_35885488/article/details/103185724
[Link 1]: http://spark.apache.org/docs/latest/streaming-programming-guide.html
[https_github.com_daizikaikou_learningSpark]: https://github.com/daizikaikou/learningSpark