# 一.DStream 整合RDD  
## 1.官网算子
![](http://img.blog.itpub.net/blog/2019/08/21/d574ed3d8ef188d4.png?x-oss-process=style/bb)
## 2.使用案例
```
生产中使用多的是一个文件中有很多域名，另一个中是黑名单，要进行剔除

数据一：日志信息    DStream
	domain,traffic
	xinlang.com
	xinlang.com
	baidu.com
数据二：已有的文件  黑名单  RDD
	domain
	baidu.com
```
## 3.RDD实现上述需求
```scala
package sparkstreaming02

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

object Demo1 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Demo1").setMaster("local[2]")
    val sc = new SparkContext(conf)
    val input1 = new ListBuffer[(String,Long)]
    input1.append(("www.xinlang.com", 8888))
    input1.append(("www.xinalng.com", 9999))
    input1.append(("www.baidu.com", 7777))
    val data1 = sc.parallelize(input1)
    //进行join一定要是key,value形式的
    val input2 = new ListBuffer[(String,Boolean)]
    input2.append(("www.baidu.com",true))
    val data2 = sc.parallelize(input2)

data1.leftOuterJoin(data2)
      .filter(x => {

x._2._2.getOrElse(false) != true

}).map(x => (x._1,x._2._1))
      .collect().foreach(println)

}

```
## 4.SparkStreaming实现
```scala
package sparkstreaming02

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.ListBuffer

object Streaming {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Streaming").setMaster("local[2]")
    val ssc = new StreamingContext(conf,Seconds(10))
    val lines = ssc.socketTextStream("s201",9999)
    // 数据二： rdd
    val input2 = new ListBuffer[(String,Boolean)]
    input2.append(("www.baidu.com",true))
    val data2 = ssc.sparkContext.parallelize(input2)

lines.map(x=>(x.split(",")(0), x)).transform(

rdd => {

rdd.leftOuterJoin(data2)
          .filter(x => {

x._2._2.getOrElse(false) != true //注意 join之后过滤

}).map(x => (x._1,x._2._1))
      }

).print()
    ssc.start()
    ssc.awaitTermination()

}

}
```
# 二.SparkStreaming插入外部数据源
## 1.插入外部数据源用的，但是使用这个有几个坑
![](http://img.blog.itpub.net/blog/2019/08/21/9f7bf6c8c45586a2.png?x-oss-process=style/bb)、
## 2.错误一官网例子
![](http://img.blog.itpub.net/blog/2019/08/21/fb94f1e41602a501.png?x-oss-process=style/bb)
## 3.原因
```
connect 在Driver端创建，record在executor，发过去序列化错误
```
![](http://img.blog.itpub.net/blog/2019/08/21/dbb2aa4afa56f99a.png?x-oss-process=style/bb)

## 4.解决
```
解决：第一种把connect放到executor端
这样弊端是每条记录会生成一个connect太耗费资源

words.foreachRDD { rdd =>
          rdd.foreach { record =>
            val connection = createConnection()  // executed at the driver
            val word = record._1
            val count = record._2.toInt
            val sql = s"insert into wc (wc,count) values($word,$count)"
           connection.createStatement().execute(sql)
         }

```
## 5.最终解决办法
```scala
package sparkstreaming02

import java.sql.DriverManager

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object MysqlStreaming {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("MysqlStreaming")
    val ssc = new StreamingContext(conf,Seconds(1))
    val lines = ssc.socketTextStream("s201",9999)
    val words = lines.flatMap(x => x.split(",")).map((_,1)).reduceByKey(_+_)

//    words.foreachRDD { rdd =>
//      val connection = createConnection()  // executed at the driver
//      rdd.foreach { record =>
//        val word = record._1
//        val count = record._2
//        val sql = s"insert into wc (word,count) values($word,$count)"
//        connection.createStatement().execute(sql)
//      }
//    }
//        words.foreachRDD { rdd =>
//          rdd.foreach { record =>
//            val connection = createConnection()  // executed at the driver
//            val word = record._1
//            val count = record._2.toInt
//            val sql = s"insert into wc (wc,count) values($word,$count)"
//            connection.createStatement().execute(sql)
//          }
//        }
    //最终的写法
    words.foreachRDD { rdd =>
      rdd.foreachPartition { partitionOfRecords =>
        val connection = createConnection()
        partitionOfRecords.foreach(
          record =>{

val word = record._1
        val count = record._2
        val sql = s"insert into wc (wc,count) values('$word',$count)"
        connection.createStatement().execute(sql)}
        )

}
    }

ssc.start()
    ssc.awaitTermination()
  }

def createConnection() = {
    Class.forName("com.mysql.cj.jdbc.Driver")
    DriverManager.getConnection("jdbc:mysql://localhost:3306/hive?serverTimezone=UTC&useSSL=false","root","123456")
  }

}
```

## 6.出现问题
```scala
错误，插入数据库时，你要插入字符串要用''

例如：
val sql = s"insert into wc (wc,count) values($word,$count)"
word是字符串，你要不加双引号就报这个错误
正确
val sql = s"insert into wc (wc,count) values('$word',$count)"

```

![](http://img.blog.itpub.net/blog/2019/08/21/ea4b96909193f76d.png?x-oss-process=style/bb)

ITPub博客

SparkStreaming 的使用与总结

最新文章