SS02 | DoubleHappy or Jepson

Transformations on DStreams

updateStateByKey：

先看一个案例

[double_happy@hadoop101 ~]$ nc -lk 9999
a,a,a,d,d
a,a,a,d,d

object StreamingWCApp01 {

  def main(args: Array[String]): Unit = {

    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)

    val lines = ssc.socketTextStream("hadoop101",9999)

    val result = lines.flatMap(_.split(","))
      .map((_,1))
      .reduceByKey(_+_)

    result.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
结果是：
-------------------------------------------
Time: 1572519050000 ms
-------------------------------------------
(d,4)
(a,6)

-------------------------------------------
Time: 1572519060000 ms
-------------------------------------------

-------------------------------------------
Time: 1572519070000 ms
-------------------------------------------

注意：
这个计算 只计算当前批次的 之后批次 没有数据 

需求：
	统计 从现在时间点 到 10分钟之后的 a出现的次数  ？对于
	上面的代码是无法满足 的    (也可以满足 存起来 再加 也可以)

对于累计的需求该这么办呢？

这就引出一个有没有状态的问题。

状态：State
    无状态的        只与当前批次有关的 叫无状态
    有状态的        前后批次是有关系的   eg：需要把之前的历史到当前的时间点 需要累计起来

实现有状态的 需求 使用updateStateByKey算子***

updateStateByKey ：更新你的状态 通过key 来更新   怎么更新 传入一个function 即可 eg:累加 还是别的 


updateStateByKey(func)	：
	Return a new "state" DStream where the state for each key is updated 
	by applying the given function on the previous state of the key 
	and the new values for the key. 
	This can be used to maintain arbitrary state data for each key.

UpdateStateByKey Operation

The updateStateByKey operation allows you to maintain arbitrary state while continuously updating it with new information. To use this, you will have to do two steps.

1.Define the state - The state can be an arbitrary data type.

2.Define the state update function - Specify with a function how to update the state
using the previous state and the new values from an input stream.

In every batch, Spark will apply the state update function for all existing keys, regardless of whether they have new data in a batch or not. If the update function returns None then the key-value pair will be eliminated.

Let’s illustrate this with an example. Say you want to maintain a running count of each word seen in a text data stream. Here, the running count is the state and it is an integer. We define the update function as:

updateStateByKey operation ：
	1.Define the state
	2.Define the state update function

对于上面给的wc例子 ：
哪个东西是state    应该是 value 

updateStateByKey  通过key 来更新谁 ( 你可以这么理解)

案例：

1. 
 val result = lines.flatMap(_.split(","))
      .map((_,1))
      .reduceByKey(_+_)
      
reduceByKey(_+_)  是对当前批次的累计 所以这里不能这么写

object StreamingWCApp01 {

  def main(args: Array[String]): Unit = {

    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)

    val lines = ssc.socketTextStream("hadoop101",9999)

    val result = lines.flatMap(_.split(","))
      .map((_,1))
      .updateStateByKey(updateFunction)

    result.print()

    ssc.start()
    ssc.awaitTermination()
  }

  /**
    *
    * 1批次：  a a a d d
    * 2批次：  b b b c c a
    *
    *newValues : 当前批次的值
    *           key对应的新值(或者有新的key)  可能有多个 所以是一个Seq
    * preValues : 以前批次的累加值
    *             key已经存在的值  有可能没有 有可能有  所以定义成Option  有就返回some  没有返回none
    *
    */
  def updateFunction(newValues: Seq[Int], preValues: Option[Int]): Option[Int] = {
    //newValues : (b,1)(b,1)(b,1)(c,1)(c,1) (a,1)

    val curr = newValues.sum // 当前批次
    val pre = preValues.getOrElse(0)   //老的值   (a,3) (d,2)   拿出值  key没有的  就赋值为0
    Some(curr + pre)
  }

}

结果：
19/10/31 19:21:07 ERROR StreamingContext: Error starting the context, marking it as stopped
java.lang.IllegalArgumentException: requirement failed: The checkpoint directory has not been set. Please set it by StreamingContext.checkpoint().
	at scala.Predef$.require(Predef.scala:224)
	at org.apache.spark.streaming.dstream.DStream.validateAtStart(DStream.scala:243)
	at org.apache.spark.streaming.dstream.DStream$$anonfun$validateAtStart$8.apply(DStream.scala:276)
	at org.apache.spark.streaming.dstream.DStream$$anonfun$validateAtStart$8.apply(DStream.scala:276)
	at scala.collection.immutable.List.foreach(List.scala:381)
	at org.apache.spark.streaming.dstream.DStream.validateAtStart(DStream.scala:276)
	at org.apache.spark.streaming.DStreamGraph$$anonfun$start$4.apply(DStreamGraph.scala:51)
	at org.apache.spark.streaming.DStreamGraph$$anonfun$start$4.apply(DStreamGraph.scala:51)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at org.apache.spark.streaming.DStreamGraph.start(DStreamGraph.scala:51)
	at org.apache.spark.streaming.scheduler.JobGenerator.startFirstTime(JobGenerator.scala:194)
	at org.apache.spark.streaming.scheduler.JobGenerator.start(JobGenerator.scala:100)
	at org.apache.spark.streaming.scheduler.JobScheduler.start(JobScheduler.scala:103)
	at org.apache.spark.streaming.StreamingContext$$anonfun$liftedTree1$1$1.apply$mcV$sp(StreamingContext.scala:583)
	at org.apache.spark.streaming.StreamingContext$$anonfun$liftedTree1$1$1.apply(StreamingContext.scala:578)
	at org.apache.spark.streaming.StreamingContext$$anonfun$liftedTree1$1$1.apply(StreamingContext.scala:578)
	at ... run in separate thread using org.apache.spark.util.ThreadUtils ... ()
	at org.apache.spark.streaming.StreamingContext.liftedTree1$1(StreamingContext.scala:578)
	at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:572)
	at com.ruozedata.spark.ss02.StreamingWCApp01$.main(StreamingWCApp01.scala:19)
	at com.ruozedata.spark.ss02.StreamingWCApp01.main(StreamingWCApp01.scala)
19/10/31 19:21:08 WARN ReceiverSupervisorImpl: Skip stopping receiver because it has not yet stared
Exception in thread "main" java.lang.IllegalArgumentException: requirement failed: The checkpoint directory has not been set. Please set it by StreamingContext.checkpoint().
	at scala.Predef$.require(Predef.scala:224)
	at org.apache.spark.streaming.dstream.DStream.validateAtStart(DStream.scala:243)
	at org.apache.spark.streaming.dstream.DStream$$anonfun$validateAtStart$8.apply(DStream.scala:276)
	at org.apache.spark.streaming.dstream.DStream$$anonfun$validateAtStart$8.apply(DStream.scala:276)
	at scala.collection.immutable.List.foreach(List.scala:381)
	at org.apache.spark.streaming.dstream.DStream.validateAtStart(DStream.scala:276)
	at org.apache.spark.streaming.DStreamGraph$$anonfun$start$4.apply(DStreamGraph.scala:51)
	at org.apache.spark.streaming.DStreamGraph$$anonfun$start$4.apply(DStreamGraph.scala:51)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48)
	at org.apache.spark.streaming.DStreamGraph.start(DStreamGraph.scala:51)
	at org.apache.spark.streaming.scheduler.JobGenerator.startFirstTime(JobGenerator.scala:194)
	at org.apache.spark.streaming.scheduler.JobGenerator.start(JobGenerator.scala:100)
	at org.apache.spark.streaming.scheduler.JobScheduler.start(JobScheduler.scala:103)
	at org.apache.spark.streaming.StreamingContext$$anonfun$liftedTree1$1$1.apply$mcV$sp(StreamingContext.scala:583)
	at org.apache.spark.streaming.StreamingContext$$anonfun$liftedTree1$1$1.apply(StreamingContext.scala:578)
	at org.apache.spark.streaming.StreamingContext$$anonfun$liftedTree1$1$1.apply(StreamingContext.scala:578)
	at ... run in separate thread using org.apache.spark.util.ThreadUtils ... ()
	at org.apache.spark.streaming.StreamingContext.liftedTree1$1(StreamingContext.scala:578)
	at org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:572)
	at com.ruozedata.spark.ss02.StreamingWCApp01$.main(StreamingWCApp01.scala:19)
	at com.ruozedata.spark.ss02.StreamingWCApp01.main(StreamingWCApp01.scala)

Process finished with exit code 1


注意：
Please set it by StreamingContext.checkpoint().

修改代码

[double_happy@hadoop101 ~]$ nc -lk 9999
a,a,b,b,a         第一次输入

a,a,b,b,a         第二次输入

object StreamingWCApp01 {

  def main(args: Array[String]): Unit = {

    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)
    
    ssc.checkpoint("file:///C:/IdeaProjects/spark/checkponit")

    val lines = ssc.socketTextStream("hadoop101",9999)

    val result = lines.flatMap(_.split(","))
      .map((_,1))
      .updateStateByKey(updateFunction)

    result.print()

    ssc.start()
    ssc.awaitTermination()
  }

  /**
    *
    * 1批次：  a a a d d
    * 2批次：  b b b c c a
    *
    *newValues : 当前批次的值
    *           key对应的新值(或者有新的key)  可能有多个 所以是一个Seq
    * preValues : 以前批次的累加值
    *             key已经存在的值  有可能没有 有可能有  所以定义成Option  有就返回some  没有返回none
    *
    */
  def updateFunction(newValues: Seq[Int], preValues: Option[Int]): Option[Int] = {
    //newValues : (b,1)(b,1)(b,1)(c,1)(c,1) (a,1)

    val curr = newValues.sum // 当前批次
    val pre = preValues.getOrElse(0)   //老的值   (a,3) (d,2)   拿出值  key没有的  就赋值为0
    Some(curr + pre)
  }

}

结果：
-------------------------------------------
Time: 1572521050000 ms
-------------------------------------------

19/10/31 19:24:13 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
19/10/31 19:24:13 WARN BlockManager: Block input-0-1572521053200 replicated to only 0 peer(s) instead of 1 peers
-------------------------------------------
Time: 1572521060000 ms
-------------------------------------------
(b,2)
(a,3)

19/10/31 19:24:25 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
19/10/31 19:24:25 WARN BlockManager: Block input-0-1572521064800 replicated to only 0 peer(s) instead of 1 peers
-------------------------------------------
Time: 1572521070000 ms
-------------------------------------------
(b,4)
(a,6)

-------------------------------------------
Time: 1572521080000 ms
-------------------------------------------
(b,4)
(a,6)


注意：
为什么要checkpoint呢？
之前的代码都是没有设置checkpoint 的 为什么之前不需要设置 呢？
因为之前的是没有状态的 没有状态 就是当前批次处理完就ok了 

但是现在 需要把当前批次 和 以前批次累加起来的  这个东西在哪里呢？下图

在这里插入图片描述

ok 现在我把程序关掉 重启以后 是多少呢？  
之前值是：
	(b,4)
    (a,6)
重启之后的值是：空的 

-------------------------------------------
Time: 1572521460000 ms
-------------------------------------------

-------------------------------------------
Time: 1572521470000 ms
-------------------------------------------

也就是说 ：
	如果你的作业 中途挂掉了 重启之后 什么都没有了 
为什么呢？
	因为之前的结果写到 checkponit里了 ，而且当前批次 也没有数据输入进来
那么：
	我们有什么办法 把 checkponit里的数据读取出来呢？
	看官网

Checkpointing

最好直接看官网：我只是截取我认为重要的
Spark Streaming needs to checkpoint enough information 
to a fault- tolerant storage system such that it can recover from failures.
 There are two types of data that are checkpointed.

1. a fault- tolerant storage system    可以选用HDFS
2. two types of data that are checkpointed
      1.Metadata checkpointing
      			Configuration     配置文件
      			DStream operations     算子 
      			Incomplete batches    未完成的
      2.Data checkpointing    就是你真正传过来的数据

When to enable Checkpointing？
   1.Usage of stateful transformations 
   2.Recovering from failures of the driver running the application 
   		driver挂了 你的作业就挂了 当你作业挂了 从Checkpoint中恢复

How to configure Checkpointing？
	看代码   就是说什么代码得改动 不能像之前那样写
1.需要定义一个函数 这个函数就是 创建StreamingContext
2.之后 再 val ssc = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _) 
才可以解决 重启之后能够拿到之前的值 

这个就是利用了 ：
	从Checkpoint中恢复 StreamingContext思想(driver 里的 )

object StreamingWCApp02 {


  val checkpointDirectory = "file:///C:/IdeaProjects/spark/checkponit"

  def main(args: Array[String]): Unit = {
    // 当作业挂了时，从checkpoint中去获取StreamingContext
    val ssc = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)
    ssc.start()
    ssc.awaitTermination()
  }

  def functionToCreateContext(): StreamingContext = {
    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)
    ssc.checkpoint(checkpointDirectory)
    val lines = ssc.socketTextStream("hadoop101",9999)
    val result = lines.flatMap(_.split(","))
      .map((_,1))
      .updateStateByKey(updateFunction)
    result.print()

    ssc
  }

  /**
    *
    * 1)  a a a d d
    * 2)  b b b c c a
    *
    * @param newValues  当前批次的值
    *        key对应的新值  可能有多个 所以是一个Seq
    * @param preValues  以前批次的累加值
    *        key已经存在的值  有可能没有 有可能有  所以定义成Option
    * @return
    */
  def updateFunction(newValues: Seq[Int], preValues: Option[Int]): Option[Int] = {
    val curr = newValues.sum // 当前
    val pre = preValues.getOrElse(0)
    Some(curr + pre)
  }
}

结果是：
	-------------------------------------------
Time: 1572521710000 ms
-------------------------------------------

-------------------------------------------
Time: 1572521720000 ms
-------------------------------------------

-------------------------------------------
Time: 1572521730000 ms
-------------------------------------------

-------------------------------------------
Time: 1572521740000 ms
-------------------------------------------

-------------------------------------------
Time: 1572521750000 ms
-------------------------------------------

-------------------------------------------
Time: 1572521760000 ms
-------------------------------------------

为什么呢？因为 我改动代码了 虽然 checkpoint目录没有变 
先把之前的 checkpoint 目录删掉 再测试 (第一次 之后关闭程序 再重启)

[double_happy@hadoop101 ~]$ nc -lk 9999
a,a,b,b,a    第一次输入 
a,a,b,b,a

a,a,b,b,a   第二次输入
a,a,b,b,a

结果：
-------------------------------------------
Time: 1572523040000 ms
-------------------------------------------
(b,4)
(a,6)

19/10/31 19:57:25 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
19/10/31 19:57:25 WARN BlockManager: Block input-0-1572523045400 replicated to only 0 peer(s) instead of 1 peers
19/10/31 19:57:26 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
19/10/31 19:57:26 WARN BlockManager: Block input-0-1572523046200 replicated to only 0 peer(s) instead of 1 peers
-------------------------------------------
Time: 1572523050000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523060000 ms
-------------------------------------------
(b,8)
(a,12)

重启后的结果：
	-------------------------------------------
Time: 1572523070000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523080000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523090000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523100000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523110000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523120000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523130000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523140000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523150000 ms
-------------------------------------------
(b,8)
(a,12)

-------------------------------------------
Time: 1572523160000 ms
-------------------------------------------
(b,8)
(a,12)

ok啦

Stream + Kafka == CP
Kafka 的offset肯定是需要手工维护：有哪些呢？很多的 
    1.checkpoint： 就是把offset维护在checkponit里面的    
    	(代码不能发生任何的变化   只要你代码发生了变化 就意味着 checkpoint 的 matadata 发生了变化  )
    2.Kafka     
    3.ZK   
    4.MySQL    
    5.Redis

所以生产上 checkpoint 根本没法用  (你的代码怎么可能不变呢？或者不修改呢？所以用不了 )

把数据写出去： ****

foreachRDD：

foreachRDD(func)：
	The most generic output operator that applies a function, func, to each RDD 
	generated from the stream. 
	This function should push the data in each RDD to an external system, 
	such as saving the RDD to files, 
	or writing it over the network to a database. Note that the function func
	 is executed in the driver process running the streaming application,
	 and will usually have RDD actions in it
	  that will force the computation of the streaming RDDs.

1. such as saving the RDD to files, 
	or writing it over the network to a database.
2.闭包  优雅的方式写出去
3.the function func
	 is executed in the driver process 
	 running the streaming application
	 func是运行在driver process的

driver端到executor端 必然涉及到一个序列化的问题

把数据写到MySQL

MySQL底层引擎有几种？各自什么区别？

在这里插入图片描述
咱们一步一步来由劣到优

object StreamingWCApp03 {

  def main(args: Array[String]): Unit = {


    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)

    //TODO... 填写我们的业务逻辑
    // Input:   socket  Input DStream
    val lines = ssc.socketTextStream("hadoop101", 9999)

    // transformation
    val result = lines.flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _)

    // output
    result.foreachRDD( rdd =>{
      val connection: Connection = MySQLUtils.getConnection()

      rdd.foreach(pair =>{
        val sql = s"insert into wc(word,cnt) values('${pair._1}', ${pair._2})"
        connection.createStatement().execute(sql)
      })

      MySQLUtils.closeResource(connection)
    })

    ssc.start()
    ssc.awaitTermination()
  }
}

结果：
19/10/31 20:43:30 ERROR JobScheduler: Error running job streaming job 1572525810000 ms.0
org.apache.spark.SparkException: Task not serializable
	at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)
	at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393)
	at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
	at org.apache.spark.SparkContext.clean(SparkContext.scala:2326)
	at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:926)
	at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:925)
	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:363)
	at org.apache.spark.rdd.RDD.foreach(RDD.scala:925)
	at com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1.apply(StreamingWCApp03.scala:32)
	at com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1.apply(StreamingWCApp03.scala:29)
	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)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
	at scala.util.Try$.apply(Try.scala:192)
	at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:257)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
	at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.NotSerializableException: java.lang.Object
Serialization stack:
	- object not serializable (class: java.lang.Object, value: java.lang.Object@4ffd7c3f)
	- writeObject data (class: java.util.HashMap)
	- object (class java.util.HashMap, {UTF-8=java.lang.Object@4ffd7c3f, US-ASCII=com.mysql.jdbc.SingleByteCharsetConverter@53c22208, WINDOWS-1252=com.mysql.jdbc.SingleByteCharsetConverter@77cd4c6d})
	- field (class: com.mysql.jdbc.ConnectionImpl, name: charsetConverterMap, type: interface java.util.Map)
	- object (class com.mysql.jdbc.JDBC4Connection, com.mysql.jdbc.JDBC4Connection@65b0d4df)
	- field (class: com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1$$anonfun$apply$1, name: connection$1, type: interface java.sql.Connection)
	- object (class com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1$$anonfun$apply$1, <function1>)
	at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
	at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
	at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
	at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:400)
	... 30 more
Exception in thread "main" org.apache.spark.SparkException: Task not serializable
	at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)
	at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:393)
	at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:162)
	at org.apache.spark.SparkContext.clean(SparkContext.scala:2326)
	at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:926)
	at org.apache.spark.rdd.RDD$$anonfun$foreach$1.apply(RDD.scala:925)
	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:363)
	at org.apache.spark.rdd.RDD.foreach(RDD.scala:925)
	at com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1.apply(StreamingWCApp03.scala:32)
	at com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1.apply(StreamingWCApp03.scala:29)
	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)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
	at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:50)
	at scala.util.Try$.apply(Try.scala:192)
	at org.apache.spark.streaming.scheduler.Job.run(Job.scala:39)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply$mcV$sp(JobScheduler.scala:257)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler$$anonfun$run$1.apply(JobScheduler.scala:257)
	at scala.util.DynamicVariable.withValue(DynamicVariable.scala:58)
	at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:256)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.NotSerializableException: java.lang.Object
Serialization stack:
	- object not serializable (class: java.lang.Object, value: java.lang.Object@4ffd7c3f)
	- writeObject data (class: java.util.HashMap)
	- object (class java.util.HashMap, {UTF-8=java.lang.Object@4ffd7c3f, US-ASCII=com.mysql.jdbc.SingleByteCharsetConverter@53c22208, WINDOWS-1252=com.mysql.jdbc.SingleByteCharsetConverter@77cd4c6d})
	- field (class: com.mysql.jdbc.ConnectionImpl, name: charsetConverterMap, type: interface java.util.Map)
	- object (class com.mysql.jdbc.JDBC4Connection, com.mysql.jdbc.JDBC4Connection@65b0d4df)
	- field (class: com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1$$anonfun$apply$1, name: connection$1, type: interface java.sql.Connection)
	- object (class com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1$$anonfun$apply$1, <function1>)
	at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
	at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
	at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
	at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:400)
	... 30 more
19/10/31 20:43:30 WARN SocketReceiver: Error receiving data
java.net.SocketException: Socket closed
	at java.net.SocketInputStream.socketRead0(Native Method)
	at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
	at java.net.SocketInputStream.read(SocketInputStream.java:171)
	at java.net.SocketInputStream.read(SocketInputStream.java:141)
	at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284)
	at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326)
	at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178)
	at java.io.InputStreamReader.read(InputStreamReader.java:184)
	at java.io.BufferedReader.fill(BufferedReader.java:161)
	at java.io.BufferedReader.readLine(BufferedReader.java:324)
	at java.io.BufferedReader.readLine(BufferedReader.java:389)
	at org.apache.spark.streaming.dstream.SocketReceiver$$anon$1.getNext(SocketInputDStream.scala:121)
	at org.apache.spark.streaming.dstream.SocketReceiver$$anon$1.getNext(SocketInputDStream.scala:119)
	at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
	at org.apache.spark.streaming.dstream.SocketReceiver.receive(SocketInputDStream.scala:91)
	at org.apache.spark.streaming.dstream.SocketReceiver$$anon$2.run(SocketInputDStream.scala:72)
19/10/31 20:43:30 WARN ReceiverSupervisorImpl: Restarting receiver with delay 2000 ms: Error receiving data
java.net.SocketException: Socket closed
	at java.net.SocketInputStream.socketRead0(Native Method)
	at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
	at java.net.SocketInputStream.read(SocketInputStream.java:171)
	at java.net.SocketInputStream.read(SocketInputStream.java:141)
	at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284)
	at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326)
	at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178)
	at java.io.InputStreamReader.read(InputStreamReader.java:184)
	at java.io.BufferedReader.fill(BufferedReader.java:161)
	at java.io.BufferedReader.readLine(BufferedReader.java:324)
	at java.io.BufferedReader.readLine(BufferedReader.java:389)
	at org.apache.spark.streaming.dstream.SocketReceiver$$anon$1.getNext(SocketInputDStream.scala:121)
	at org.apache.spark.streaming.dstream.SocketReceiver$$anon$1.getNext(SocketInputDStream.scala:119)
	at org.apache.spark.util.NextIterator.hasNext(NextIterator.scala:73)
	at org.apache.spark.streaming.dstream.SocketReceiver.receive(SocketInputDStream.scala:91)
	at org.apache.spark.streaming.dstream.SocketReceiver$$anon$2.run(SocketInputDStream.scala:72)
19/10/31 20:43:30 ERROR ReceiverTracker: Deregistered receiver for stream 0: Stopped by driver
19/10/31 20:43:30 WARN ReceiverSupervisorImpl: Receiver has been stopped
Exception in thread "receiver-supervisor-future-0" java.lang.Error: java.lang.InterruptedException: sleep interrupted
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1155)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.InterruptedException: sleep interrupted
	at java.lang.Thread.sleep(Native Method)
	at org.apache.spark.streaming.receiver.ReceiverSupervisor$$anonfun$restartReceiver$1.apply$mcV$sp(ReceiverSupervisor.scala:196)
	at org.apache.spark.streaming.receiver.ReceiverSupervisor$$anonfun$restartReceiver$1.apply(ReceiverSupervisor.scala:189)
	at org.apache.spark.streaming.receiver.ReceiverSupervisor$$anonfun$restartReceiver$1.apply(ReceiverSupervisor.scala:189)
	at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)
	at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	... 2 more

注意：
	org.apache.spark.SparkException: Task not serializable
	ClosureCleaner    Closure 闭包的意思
根本原因是：
	Caused by: java.io.NotSerializableException: java.lang.Object
Serialization stack:
	- object not serializable (class: java.lang.Object, value: java.lang.Object@4ffd7c3f)
	- writeObject data (class: java.util.HashMap)
	- object (class java.util.HashMap, {UTF-8=java.lang.Object@4ffd7c3f, US-ASCII=com.mysql.jdbc.SingleByteCharsetConverter@53c22208, WINDOWS-1252=com.mysql.jdbc.SingleByteCharsetConverter@77cd4c6d})
	- field (class: com.mysql.jdbc.ConnectionImpl, name: charsetConverterMap, type: interface java.util.Map)
	- object (class com.mysql.jdbc.JDBC4Connection, com.mysql.jdbc.JDBC4Connection@65b0d4df)
	- field (class: com.ruozedata.spark.ss02.StreamingWCApp03$$anonfun$main$1$$anonfun$apply$1, name: connection$1, type: interface java.sql.Connection)

就是 object not serializable ：com.mysql.jdbc.SingleByteCharsetConverter 
MySQL的驱动不能序列化    但是事实上 MySQL驱动就是序列化不了 

该怎么办呢？ 看官网  下图

在这里插入图片描述
华丽的分割线————————————————————————————————————

上面的错误明白之后 那么什么叫做闭包？
先看一下官网  RDD篇介绍的

Understanding closures

闭包：在函数内部 引用了一个外部的变量 
eg： 这两行代码

     val connection: Connection = MySQLUtils.getConnection()
      rdd.foreach(pair =>{
        val sql = s"insert into wc(word,cnt) values('${pair._1}', ${pair._2})"
        connection.createStatement().execute(sql)
      })

foreach 内部使用了 connection  而connection 是在foreach的外部

如果 假设哈 connection 可以序列化 的  上面这种写法是没有问题的！！！
很不幸 connection objects are rarely transferable across machines

修改：

    result.foreachRDD( rdd =>{
      rdd.foreach(pair =>{
        val connection: Connection = MySQLUtils.getConnection()
        val sql = s"insert into wc(word,cnt) values('${pair._1}', ${pair._2})"
        connection.createStatement().execute(sql)
        MySQLUtils.closeResource(connection)
      })
    })

connection放到里面去  那么还涉及闭包的问题么？
一定没有闭包的问题了 避免了上次测试 出现的闭包问题

运行结果：
没有日志的 因为 foreachRDD 是没有返回值的  只能查看MySQL数据了

-------------------------------------------------------------------------------------------
[double_happy@hadoop101 ~]$ nc -lk 9999
a,a,b,b,a
a,a,b,b,a
a,a,b,b,a
a,a,b,b,a
--------------------------------------------------------------------------------------------
mysql> select * from wc;
Empty set (0.00 sec)

mysql> select * from wc;
+------+------+
| word | cnt  |
+------+------+
| b    |    8 |
| a    |   12 |
+------+------+
2 rows in set (0.00 sec)

mysql> 

说明写也是ok的 

但是也有个问题的？

在这里插入图片描述

优化

object StreamingWCApp03 {

  def main(args: Array[String]): Unit = {

    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)

    //TODO... 填写我们的业务逻辑
    // Input:   socket  Input DStream
    val lines = ssc.socketTextStream("hadoop101", 9999)

    // transformation
    val result = lines.flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _)

    // output
    result.foreachRDD(rdd => {
      rdd.foreachPartition(partition => {
        val connection: Connection = MySQLUtils.getConnection()

        partition.foreach(pair => {
          val sql = s"insert into wc(word,cnt) values('${pair._1}', ${pair._2})"
          connection.createStatement().execute(sql)
        })
        MySQLUtils.closeResource(connection)
      })
    })

    ssc.start()
    ssc.awaitTermination()
  }
}

结果是：
[double_happy@hadoop101 ~]$ nc -lk 9999
a,a,b,b,a
a,a,b,b,a

mysql> select * from wc;
+------+------+
| word | cnt  |
+------+------+
| b    |    8 |
| a    |   12 |
| b    |    4 |
| a    |    6 |
+------+------+
4 rows in set (0.00 sec)

mysql> 

这种方式比前面的好多了 但是也不行 
分区多了  connection也会多  
那么最好的方式是什么呢？拿一个连接池 用完之后返回回去

在这里插入图片描述

正确的写法会写了 但是 
mysql> select * from wc;
+------+------+
| word | cnt  |
+------+------+
| b    |    8 |
| a    |   12 |
| b    |    4 |
| a    |    6 |
+------+------+
4 rows in set (0.00 sec)

mysql> 

结果咋整 写了两次就这样了  怎么解决呢？这是数据问题

还有一种写法建议使用它

scalikejdbc 自带Connection Pool
在这里插入图片描述

object StreamingWCApp03 {

  def main(args: Array[String]): Unit = {


    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)

    //TODO... 填写我们的业务逻辑
    // Input:   socket  Input DStream
    val lines = ssc.socketTextStream("hadoop101", 9999)


    // transformation
    val result = lines.flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _)

    // output
    
    DBs.setupAll() //这样就把配置文件解析出来了
    result.foreachRDD(rdd => {

      rdd.foreachPartition(partition => {
      
        partition.foreach(pair => {
          DB.autoCommit { implicit session => {
            SQL("insert into wc(word,cnt) values(?, ?)")
              .bind(pair._1,pair._2)
              .update().apply()
          }
          }
        })
      })
    })

    ssc.start()
    ssc.awaitTermination()
  }
}

结果：
mysql> select * from wc;
+------+------+
| word | cnt  |
+------+------+
| b    |   20 |
| a    |   30 |
+------+------+
2 rows in set (0.00 sec)

mysql> 


注意：
你确定scalikejdbc 默认就使用 连接池么？？？ 留一个坑

之前 我们用state 进行累计的  
因为用state累加 会用到checkpoint   checkpoint自己生成小文件一大堆  等等

那么 不用state 能不能累加？
用redis 
 /**
      * WC这种统计维度来说
      * Redis的使用关键点：如何选择合适的数据类型
      */
这里我们选hash

object StreamingWCApp03 {

  def main(args: Array[String]): Unit = {

    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)

    //TODO... 填写我们的业务逻辑
    // Input:   socket  Input DStream
    val lines = ssc.socketTextStream("hadoop101", 9999)


    // transformation
    val result = lines.flatMap(_.split(",")).map((_, 1)).reduceByKey(_ + _)

    // output
    /**
      * WC这种统计维度来说
      * Redis的使用关键点：如何选择合适的数据类型
      */
        result.foreachRDD(rdd => {
          rdd.foreachPartition(partition => {
            val jedis = RedisUtils.getJedis  // 获取Redis连接
            partition.foreach(pair => {
              jedis.hincrBy("doublehappy_redis_wc", pair._1, pair._2)   //String key, String field, long value
            })
            jedis.close() // free
          })
        })

    ssc.start()
    ssc.awaitTermination()
  }
}

扩展：
	这里是连接redis  ，那么连接 phoneix 、Cassandra   都一样的 

结果：
hadoop101:6379> keys *
1) "name"
2) "doublehappy_redis_wc"
hadoop101:6379>

在这里插入图片描述

再放一些数据
[double_happy@hadoop101 ~]$ nc -lk 9999
a,a,b,b,a
a,a,b,b,a
a,a,b,b,a

在这里插入图片描述
说明结果ok的哈

transform
transform

transform(func)	;
Return a new DStream by applying a RDD-to-RDD function 
to every RDD of the source DStream.
 This can be used to do arbitrary RDD operations on the DStream.

之前的编程都是基于DStream
    /**
      * 现在的编程都是基于DStream    生产上绝大多数是DStream
      *
      但是 
      * DStream与RDD互操作咋整？ 使用transform
      */

好处就是 把DStream  的RDD 跟我们的RDD进行操作

需求：
     * 流处理的时候，有一个数据来源于文本或者是其他的   这部分数据是 RDD
      * 另外一个数据是来自Kafka、或者其他的数据源 这部分数据是 DStream
      
做这两个关联  你需要用到 transform

例子
黑名单
目的：
只要由黑名单里的东西把黑名单的数据全部过滤掉

先用core的方式;
object CoreBlackListApp {

  def main(args: Array[String]): Unit = {

    val sc = ContextUtils.getSparkContext("CoreBlackListApp")

    /**
      * 构建黑名单  (xx, true) 或者  (xx, 1)
      */
    val blacks = new ListBuffer[(String,Boolean)]()
    blacks.append(("苍老师",true))  // 鉴黄
    val blacksRDD = sc.parallelize(blacks)  // 把数据转成RDD

    /**
      * 构建访问日志
      */
    val input = new ListBuffer[(String,String)]
    input.append(("历史第一人","被小卡干了，000000"))
    input.append(("日天","也被小卡干了，111111"))
    input.append(("苍老师","我们敬爱的老师，111111"))
    val inputRDD = sc.parallelize(input)

    //TODO... 想从访问日志中过滤掉“苍老师”的数据
    inputRDD.leftOuterJoin(blacksRDD)
      .filter(_._2._2.getOrElse(false) != true)
      .map(x =>(x._1, x._2._1))
      .printInfo()


    sc.stop()
  }


}

结果是：
(日天,也被小卡干了，111111)
(历史第一人,被小卡干了，000000)
-------------------------

ssc：很重要 
生产上用的很多  生产上统计结果有些数据 有些是MySQL里的直接拿的 

object StreamingWCApp03 {

  def main(args: Array[String]): Unit = {

    val ssc = ContextUtils.getStreamingContext(this.getClass.getSimpleName, 10)

    //TODO... 填写我们的业务逻辑
    // Input:   socket  Input DStream
    val lines = ssc.socketTextStream("hadoop101", 9999)

    /**
      * 构建黑名单  (xx, true)  (xx, 1)
      */
    val blacks = new ListBuffer[(String,Boolean)]()
    blacks.append(("canglaoshi",true))  // 鉴黄
    val blacksRDD = ssc.sparkContext.parallelize(blacks)  // 把数据转成RDD

    // "日天","也被小卡干了，111111"
    lines.map(x => (x.split(",")(0), x))
      .transform(rdd => {
        rdd.leftOuterJoin(blacksRDD)
          .filter(_._2._2.getOrElse(false) != true)
          .map(x=>x._2._1)
      }).print()

    ssc.start()
    ssc.awaitTermination()
  }
}

结果是：
[double_happy@hadoop101 ~]$ nc -lk 9999
canglaoshi,xxooll
longlaoshi,11oooxxx
james,xxxxx

-------------------------------------------
Time: 1572533110000 ms
-------------------------------------------

-------------------------------------------
Time: 1572533120000 ms
-------------------------------------------
james,xxxxx
longlaoshi,11oooxxx

-------------------------------------------
Time: 1572533130000 ms
-------------------------------------------


结果正确 过滤掉 canglaoshi的数据