sparkStraming存储数据到mysql
Posted 伊米伊念
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了sparkStraming存储数据到mysql相关的知识,希望对你有一定的参考价值。
package sparkStreaming import org.apache.spark.SparkConf import org.apache.spark.streaming.Seconds import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.kafka.KafkaUtils import org.apache.spark.HashPartitioner import org.apache.spark.streaming.Duration import org.apache.spark.sql.SQLContext import org.apache.spark.{SparkContext, SparkConf} import spark.bean.orders import java.util.Properties import java.sql.{DriverManager, PreparedStatement, Connection} import org.apache.spark.{SparkContext, SparkConf} object WebPagePopularityValueCalculator { def main(args: Array[String]) { val Array(zkQuorum, group, topics) = Array("localhost:2181", "1", "sun_test_topic") val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]") val ssc = new StreamingContext(sparkConf, Seconds(2)) ssc.checkpoint("checkpoint") val topicpMap = topics.split(",").map((_, 2)).toMap val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicpMap).map(_._2) val popularityData = lines.map { msgLine => { val dataArr: Array[String] = msgLine.split("\\|") val pageID = dataArr(0) val popValue: Double = dataArr(1).toFloat * 0.8 + dataArr(2).toFloat * 0.8 + dataArr(3).toFloat * 1 (pageID, popValue) } } //sum the previous popularity value and current value val updatePopularityValue = (iterator: Iterator[(String, Seq[Double], Option[Double])]) => { iterator.flatMap(t => { val newValue: Double = t._2.sum val stateValue: Double = t._3.getOrElse(0); Some(newValue + stateValue) }.map(sumedValue => (t._1, sumedValue))) } val initialRDD = ssc.sparkContext.parallelize(List(("page1", 0.00))) val stateDstream = popularityData.updateStateByKey[Double](updatePopularityValue, new HashPartitioner(ssc.sparkContext.defaultParallelism), true, initialRDD) //set the checkpoint interval to avoid too frequently data checkpoint which may //may significantly reduce operation throughput stateDstream.checkpoint(Duration(8 * 2 * 1000)) //after calculation, we need to sort the result and only show the top 10 hot pages stateDstream.foreachRDD { rdd => { val sortedData = rdd.map { case (k, v) => (v, k) }.sortByKey(false) val topKData = sortedData.take(10).map { case (v, k) => (k, v) } topKData.foreach{ case (k, v) => if(v != 0) { println("page" + k + " " + "value" + v) val itb = Iterator((k, v)) tomysql(itb) } } } } ssc.start() ssc.awaitTermination() } def toMySql(iterator: Iterator[(String, Double)]): Unit = { var conn: Connection = null var ps: PreparedStatement = null val sql = "insert into userbehavior(page, number) values (?, ?)" try { Class.forName("com.mysql.jdbc.Driver"); conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/spark", "root", "[email protected]") iterator.foreach(dataIn => { ps = conn.prepareStatement(sql) ps.setString(1, dataIn._1) ps.setDouble(2, dataIn._2) ps.executeUpdate() } ) } catch { case e: Exception => e.printStackTrace() } finally { if (ps != null) { ps.close() } if (conn != null) { conn.close() } } } }
以上是关于sparkStraming存储数据到mysql的主要内容,如果未能解决你的问题,请参考以下文章
入门大数据---Spark_Streaming整合Flume