1.8.7 大数据-Spark-SparkStreaming实时流处理（保存到Mysql）

本文主要是介绍1.8.7 大数据-Spark-SparkStreaming实时流处理（保存到Mysql），希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

演练环境搭建

安装nc 作为输出流

[kfk@bigdata-pro03 softwares]$ sudo rpm -ivh nc-1.84-22.el6.x86_64.rpm Preparing...                                                            (100%########################################### [100%]1:nc                                                                 ( 19%########################################### [100%]
[kfk@bigdata-pro03 softwares]$ which nc
/usr/bin/nc

$ nc -lk 9999

官网演示DEMO

$ ./bin/run-example streaming.NetworkWordCount localhost 9999

spark-shell演示

[kfk@bigdata-pro03 spark-2.2.0-bin]$ bin/spark-shell
20/06/24 00:39:23 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
20/06/24 00:39:23 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
Spark context Web UI available at http://192.168.0.153:4041
Spark context available as 'sc' (master = local[*], app id = local-1592973563887).
Spark session available as 'spark'.
Welcome to____              __/ __/__  ___ _____/ /___\ \/ _ \/ _ `/ __/  '_//___/ .__/\_,_/_/ /_/\_\   version 2.2.0/_/Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_11)
Type in expressions to have them evaluated.
Type :help for more information.scala> import org.apache.spark._
import org.apache.spark._scala> import org.apache.spark.streaming._
import org.apache.spark.streaming._scala> val ssc = new StreamingContext(sc, Seconds(5))
ssc: org.apache.spark.streaming.StreamingContext = org.apache.spark.streaming.StreamingContext@1002b06dscala> val lines = ssc.socketTextStream("localhost", 9999)
lines: org.apache.spark.streaming.dstream.ReceiverInputDStream[String] = org.apache.spark.streaming.dstream.SocketInputDStream@514f2020scala> val words = lines.flatMap(_.split(" "))
words: org.apache.spark.streaming.dstream.DStream[String] = org.apache.spark.streaming.dstream.FlatMappedDStream@4f5df012scala> val pairs = words.map(word => (word, 1))
pairs: org.apache.spark.streaming.dstream.DStream[(String, Int)] = org.apache.spark.streaming.dstream.MappedDStream@39f3285dscala> val wordCounts = pairs.reduceByKey(_ + _)
wordCounts: org.apache.spark.streaming.dstream.DStream[(String, Int)] = org.apache.spark.streaming.dstream.ShuffledDStream@3299e315scala> wordCounts.print()scala> ssc.start()  
//一直运行，除非人为干预再停止 ssc.awaitTermination()

在nc下输入单词在shell客户端就可以读到了

在IDEA中代码

包含读到MySQL库、注释部分读到HDFS

package com.spark.streamingimport java.sql.DriverManagerimport org.apache.spark._
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.{Seconds, StreamingContext}object TestStreaming {def main(args: Array[String]): Unit = {val spark = SparkSession.builder.master("local[2]").appName("straming").getOrCreate()val sc = spark.sparkContext;sc.setLogLevel("WARN");val ssc = new StreamingContext(sc, Seconds(5))val lines = ssc.socketTextStream("bigdata-pro03.kfk.com", 9999)val words = lines.flatMap(_.split(" ")).map(word => (word, 1)).reduceByKey(_ + _)//words.foreachRDD(wd =>wd.saveAsTextFile("hdfs://bigdata-pro01.kfk.com/user/kfk/stream"));words.foreachRDD(rdd=>rdd.foreachPartition(line =>{Class.forName("com.mysql.jdbc.Driver")val conn = DriverManager.getConnection("jdbc:mysql://bigdata-pro01.kfk.com/test","root","123456")try{for (row <- line){val sql = "insert into webCount(titleName,count) values ('"+row._1+"',"+row._2+" )";conn.prepareStatement(sql).executeUpdate()}}finally {conn.close()}}))words.print()ssc.start()ssc.awaitTermination()}
}