Spark Streaming 简介_4.使用flume作为spark streaming数据源,统计每个时间片内rdd元素的个数。

1. 流计算概述

1.1 流计算简介

        数据仓库中存放的大量历史数据就是静态数据，可以利用数据挖掘和OLAP分析工具从静态数据中找到对企业有价值的信息。而流数据表现为数据以大量、快速、时变的流形式持续到达。如PM2.5检测、电子商务网站用户点击流。

1. 数据快速持续到达，潜在大小也许是无穷无尽的；

2. 数据来源众多，格式复杂；

3. 注重数据的整体价值，不过分关注个别数据；

4. 数据量大，但是不十分关注存储，一旦经过处理，要么被丢弃，要么被归档存储；

5. 数据顺序颠倒，或者不完整，系统无法控制将要处理的新到达的数据元素的顺序。

 批量计算和实时计算

 流计算： 实时获取来自不同数据源的海量数据，经过实时分析处理，获得有价值的信息。数据价值随着时间流逝而降低(如用户点击流)，因此当事件出现时就应该立即处理，而不是缓存起来进行批量处理。

流计算系统要求：高性能、海量式、实时性、分布式、易用性、可靠性

1.2 流计算与Hadoop

Hadoop设计的初衷是面向大规模数据的批量处理，每台机器并行运行MR任务，最后对结果进行汇总输出。其中MR是专门面向静态数据的批量处理的，内部各种实现机制都为批处理做了高度优化，不适合用于处理持续到达的动态数据。

降低批处理时间延迟，改造MR实现近似流计算

1. 采用变通的方式对MR进行相关的改造；

2. MR的批量数据转化成很多的小量数据，一大批数据把它切割成很多小批；

3. 每隔一个周期去启动一次MR作业，可以变相地完成一个流式数据的处理。

问题：

1. 切分成小片段，可以降低延迟，但也增加了附加开销，还要处理片段之间的依赖关系；

2. 需要改造MR以支持流式数据。

结论： Hadoop擅长批处理，不适合流计算！

其他常见流计算框架：Twitter Storm、Yahoo! S4、IBM StreamBase、IBM InfoSphere Streams等

1.3 流计算处理流程

1. 数据实时采集

通常采集多个数据源的海量数据，需要保证实时性、低延迟和稳定可靠。

 常见开源分布式日志采集系统：kafaka、flume、scribe

2. 数据实时计算

对采集的数据进行实时的分析和计算并反馈实时结果。经流处理系统处理后的数据，可视情况进行存储，以便之后再进行分析计算；在失效性要求较高的场景中，处理之后的数据也可以直接丢弃。

 3. 实时查询服务

        经过流计算框架得出的结果，可以让用户能够进行实时的查询展示和存储。另外，流处理计算结果会不断地去更新、不断地实时推送给用户，而传统的数据处理方式，需要用户主动发出查询，且获取的都是过去某一个历史时刻的快照。

2. Spark Streaming

2.1 Spark Streaming 的设计

可以整合多种输入源，下图为其支持的输入、输出数据源。

 Spark Streaming的执行流程：

       其基本原理是将实时输入数据流以时间片(秒级)为单位进行拆分，然后经Spark引擎以类似批处理的方式处理每个时间片数据。

 DStream 操作示意图

       Spark Streaming 最主要的抽象是DStream(Discretized Stream，离散化数据流)，表示连续不断的数据流。在内部实现上，Spark Streaming的输入数据按照时间片(如1秒)分成一段一段的DStream，每一段数据转换为Spark中的RDD，并且对DStream的操作都最终转变为对应的RDD操作。

2.2 Spark Streaming与Storm的对比 

Spark Streaming无法实现毫秒级的流计算，而Storm可以实现毫秒级响应。

1. Spark的低延迟执行(100ms+)，可以用于实时计算；

2. 相比于Storm，RDD数据集更容易做高效的容错处理。

 Spark Streaming采用的小批量处理的方式使得它可以同时兼容批量和实时数据处理的逻辑和算法，因此，方便了一些需要历史数据和实时数据联合分析的特定应用场合。

3. DStream操作概述

3.1 Spark Streaming工作原理

 1. 在Spark Streaming中，会有一个组件Receiver，作为一个长期运行的task跑在一个Executor上；

2. 每个Receiver都会负责一个input DStream（比如从文件中读取数据的文件流、套接字流、或从Kafka中读取的一个输入流等）

3. Spark Streaming通过input DStream与外部数据源进行连接，读取相关数据。

3.2 Spark Streaming程序基本步骤

1. 创建输入DStream来定义输入源；

2. 对DStream应用转换操作和输出操作来定义流计算；

3. 用streamingContext.start()来开始接收数据和处理流程；

4. 通过streamingContext.awaitTermination()方法来等待处理结束；

5. 通过streamingContext.stop()来手动结束流计算进程。

3.2 创建StreamingContext对象

1. 首先要生成一个StreamingContext对象，它是Spark Streaming程序的主入口；

2. 可以从SparkConf对象创建一个StreamingContext对象；

3. spark-shell提供了一个默认的SparkConext，即sc，可用以创建StreamingContext对象。

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
 
val ssc = new StreamingContext(spark.sparkContext, Seconds(1))

4. 输入源

4.1 文件流

新建一个待监控的目录  /usr/local/spark/mycode/steaming/logs，只捕捉动态变化的部分。

file1.txt

I love Hadoop
I love Spark

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
 
object Test {
    def main(args: Array[String]): Unit = {
        val spark = SparkSession
          .builder()
          .appName("wc")
          .master("local")
          .getOrCreate()
        spark.sparkContext.setLogLevel("ERROR")
        val input_path = "/Users/zz/Desktop/input/"
        val ssc = new StreamingContext(spark.sparkContext, Seconds(10))
        val lines = ssc.textFileStream(input_path)
        val wc_rdd = lines.flatMap(_.split(" "))
          .map(x => (x,1))
          .reduceByKey(_+_)
        wc_rdd.print()
        ssc.start()  // 进入循环监听状态
        ssc.awaitTermination()  // 阻塞，等待关闭
        spark.stop()
    }
}

在监控目录下新建一个文件file2.txt，就可以在舰艇窗口中显示词频统计结果。

 

打包编译运行

/usr/local/spark/bin/spark-submit --class "WCStreaming" /usr/local/spark/mycode/streaming/target/scala-2.11/simple-project_2.11-1.0.jar

执行后就进入了监听状态。

4.2 套接字流

4.2.1 使用nc程序产生套接字

监听Socket端口，接收数据

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming._
import org.apache.spark._
import org.apache.spark.storage.StorageLevel
 
object Test {
    def main(args: Array[String]): Unit = {
        if(args.length < 2){
            System.err.println("Usage: Test <hostname> <port>")
            System.exit(1)
        }
        val host = args(0)
        val port = args(1)
        StreamingExamples.setStreamingLogLevels()
        val spark = SparkSession
          .builder()
          .appName("wc")
          .master("local")
          .getOrCreate()
        // spark.sparkContext.setLogLevel("ERROR")
        val ssc = new StreamingContext(spark.sparkContext, Seconds(1))
        val lines = ssc.socketTextStream(host, port.toInt, StorageLevel.MEMORY_AND_DISK_SER)
        val wc_rdd = lines.flatMap(_.split(" "))
          .map(x => (x,1))
          .reduceByKey(_+_)
        wc_rdd.print()
        ssc.start()
        ssc.awaitTermination()
        spark.stop()
    }
}

在相同目录下新建一个StreamingExamples.scala，作为日志格式化文件。

import org.apache.spark.internal.Logging
import org.apache.log4j.{Level, Logger}
 
object StreamingExamples extends Logging {
  def setStreamingLogLevels(): Unit ={
    val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
    if(!log4jInitialized){
      logInfo("Setting log level to [WARN] for streaming example.")
      Logger.getRootLogger.setLevel(Level.WARN)
    }
  }
}

编译运行

/usr/local/spark/bin/spark-submit \
  --class "Test" \
  /usr/local/spark/mycode/streaming/target/scala-2.11/simple-project_2.11-1.0.jar \
  localhost 9999

新开一个窗口作为nc窗口，启动nc程序

nc -lk 9999

在nc端口输入一些单词，就会被监听到。

4.2.2 自己编写程序产生Socket数据源

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming._
import org.apache.spark._
import org.apache.spark.storage.StorageLevel
import java.io.PrintWriter
import java.net.ServerSocket
import scala.io.Source
 
object DataSourceSocket {
    def main(args: Array[String]): Unit = {
        def index(length: Int)={
            val rdm = new java.util.Random()
            rdm.nextInt(length)
        }
        if(args.length != 3){
            System.err.println("Usage: <filename> <port> <millisecond>")
            System.exit(1)
        }
        val fileName = args(0)
        val port = args(1)
        val millisecond = args(2)
        val lines = Source.fromFile(fileName).getLines().toList
        val rowCount = lines.length
        
        val listener = new ServerSocket(port.toInt)
        while(true){
            val socket = listener.accept()
            new Thread(){
                override def run={
                    println("Got client connected from: " + socket.getInetAddress)
                    val out = new PrintWriter(socket.getOutputStream, true)
                    while(true){
                        Thread.sleep(millisecond.toLong)
                        val context = lines(index(rowCount))
                        println(context)
                        out.write(context + "\n")
                        out.flush()
                    }
                    socket.close()
                }
            }.start()
        }
    }
}

4.3 RDD队列流

使用streamingContext.queueStream(queueOfRDD)创建基于RDD队列的DStream。

实现每隔1s创建一个RDD，Streaming每隔2s就对数据进行处理。

package com.khj.spark
 
import org.apache.spark.sql.SparkSession
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import scala.collection.mutable.SynchronizedQueue
 
object QueueStream {
    def main(args: Array[String]): Unit = {
        val spark = SparkSession
          .builder()
          .appName("wc")
          .master("local")
          .getOrCreate()
        val ssc = new StreamingContext(spark.sparkContext, Seconds(2))
        val rddQueue = new SynchronizedQueue[RDD[Int]]()
        val wc_rdd = ssc.queueStream(rddQueue)
          .map(r => (r%10, 1))
          .reduceByKey(_+_)
        wc_rdd.print()
        ssc.start()
        for(i <- 1 to 10){
            rddQueue += ssc.sparkContext.makeRDD(1 to 100, 2)
            Thread.sleep(1000)
        }
        ssc.stop()
        spark.stop()
    }
}

4.4 高级数据源

4.4.1 Apache Kafka

参考：  Spark Streaming 读取Kafka数据源

4.4.2 Apache Flume

暂无

5. 转换操作

5.1 DStream无状态转换操作

1. map(func)

对源DStream的每个元素，采用func()进行转换，得到一个新的DStream；

2. flatMap(func)

与map类似，但每个输入项可用被映射为0或多个输出项；

3. filter(func)

返回一个新的DStream，仅包含源DStream中满足func()的项；

4. repartition(num)

通过创建更多或者更少的分区改变DStream的并行程度。

5. reduce(func)

利用func()聚集源DStream中每个RDD的元素，返回一个包含单元素RDDs的新DStream。

6. count()

统计源DStream中每个RDD的元素数量。

7. union(otherStream)

返回一个新的DStream，包含源DStream和其他DStream的元素。

8. countByValue()

应用于元素类型为K的DStream上，返回一个kv类型的新DStream，每个键的值是在原DStream的每个RDD中出现的次数。

9. reduceByValue(func, [numTasks])

返回的DStream中，每一个key的值均由给定的reduce(func)聚集起来。

10. join(otherStream, [numTasks])

应用于两个DStream（一个为kv，一个为kw），返回一个包含(k,(v,w))的新DStream。

11. cogroup(otherStream, [numTasks])

应用于两个DStream（一个为kv，一个为kw），返回一个包含(k,Seq[V],Seq[w])的元组。

12. transform(func)

通过对源DStream的每个RDD应用RDD-to-RDD函数，创建一个新的DStream。支持在新的DStream中做任何RDD操作。

无状态转换操作实例

        之前“套接字流”部分介绍的词频统计，采用的就是无状态转换，每次统计都是只统计当前批次到达的单词的词频，和之前批次无关，不会进行累计。

5.2 DStream有状态转换操作

5.2.1 滑动窗口转换操作

1. 事先设定一个滑动窗口的长度(即窗口的持续时间)；

2. 设定滑动窗口的时间间隔(每隔多长时间执行一次计算)，让窗口按照指定时间间隔在源DStream上滑动；

3. 每次窗口停放的位置上，都会有一部分DStream(或者一部分RDD)被框入窗口内，形成一个小段的DStream；

4. 可以启动对这个小段DStream的计算。

 一些常见窗口转换操作：

1. window(windowLength, slideInterval)

基于源DStream产生的窗口化的批数据，计算得到一个新的DStream。

2. countByWindow(windowLength, slideInterval)

返回流中元素的一个滑动窗口数。

3. reduceByWindow(func, windowLength, slideInterval)

返回一个单元素流，利用func()聚集滑动时间间隔的流的元素创建这个单元素流。func()必须满足结合律，从而可以支持并行计算。

4. reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])

        应用到一个kv组成的DStream上时，会返回一个由kv组成的新的DStream。每一个key的值均由给定的func()进行聚合计算。

        注意： 在默认情况下，这个算子利用了Spark默认的并发任务数去分组。可以通过numTasks参数的设置来指定不同的任务数。

5. reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])

        更加高效的 reduceByKeyAndWindow，每个窗口的reduce值，是基于先前窗口的reduce值进行增量计算得到的；它会对进入滑动窗口的新数据进行reduce操作，并对离开窗口的老数据进行“逆向reduce”操作。但只能用于“可逆reduce函数”，即那些reduce函数都有一个对应的“逆向reduce函数”invFunc。

val wordCounts = pair.reduceByKeyAndWindow(_+_, _-_, Minutes(2), Seconds(10), 2)

5.2.2 updateStateByKey操作

应用场景： 在跨批次之间维护状态时。

在词频统计实例中，对于有状态转换操作而言，本批次的词频统计，会在之前批次的词频统计结果的基础上进行不断累加，所以，最终得到的词频，是所有批次的单词的总词频统计结果。

NetworkWordCountStateful.scala

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
 
object NetworkWordCountStateful {
  // 定义状态更新函数
  val updateFunc = (values: Seq[Int], state: Option[Int]) => {
    val currentCount = values.fold(0)(_+_)
    val previousCount = state.getOrElse(0)
    Some(currentCount + previousCount)
  }
 
  def main(args: Array[String]): Unit = {
    // 自定义类，设置log4j日志级别
    StreamingExamples.setStreamingLogLevels()
    val spark = SparkSession
      .builder()
      .appName("wc")
      .master("local")
      .getOrCreate()
    val ssc = new StreamingContext(spark.sparkContext, Seconds(5))
    // 设置检查点，检查点具有容错机制
    ssc.checkpoint("file:///usr/local/spark/mycode/streaming/stateful")
    val lines = ssc.socketTextStream("localhost", 9999)
    val stateDStream = lines.flatMap(_.split(" "))
      .map(x => (x,1))
      .updateStateByKey[Int](updateFunc)
    stateDStream.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

6. DStream输出操作

6.1 输出到文本文件

import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
 
object KhjDemo {
    def main(args: Array[String]): Unit = {
        StreamingExamples.setStreamingLogLevels()
        val inputPath = "/user/zz/Desktop/checkpoint"
        val outputPath = "/user/zz/Desktop/output"
        val spark = SparkSession
          .builder()
          .appName("wc")
          .master("local")
          .getOrCreate()
        // 定义状态更新函数
        val updateFunc = (values: Seq[Int], state: Option[Int]) => {
            val currentCount = values.fold(0)(_+_)
            val previousCount = state.getOrElse(0)
            Some(currentCount + previousCount)
        }
        val ssc = new StreamingContext(spark.sparkContext, Seconds(5))
        ssc.checkpoint(inputPath)
        val lines = ssc.socketTextStream("localhost", 9999)
        val stateDStream = lines.flatMap(_.split(" "))
          .map((_,1))
          .updateStateByKey[Int](updateFunc)
        // stateDStream.print()
        // 写入到文本文件中
        stateDStream.saveAsTextFiles(outputPath)
        
        ssc.start()
        ssc.awaitTermination()  // 等待终止
        spark.stop()
    }
}

6.2 写入MySQL

在sprak数据库中创建一个名为wordcount的表。

use spark

create table wordcount(word char(20), count int(4))

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds
import java.sql.{PreparedStatement, Connection, DriverManager}
 
object Test {
    def main(args: Array[String]): Unit = {
        StreamingExamples.setStreamingLogLevels()
        val inputPath = "/user/zz/Desktop/checkpoint"
        val spark = SparkSession
          .builder()
          .appName("wc")
          .master("local")
          .getOrCreate()
        // 定义状态更新函数
        val updateFunc = (values: Seq[Int], state: Option[Int]) => {
            val currentCount = values.fold(0)(_+_)
            val previousCount = state.getOrElse(0)
            Some(currentCount + previousCount)
        }
        val ssc = new StreamingContext(spark.sparkContext, Seconds(5))
        ssc.checkpoint(inputPath)
        val lines = ssc.socketTextStream("localhost", 9999)
        val stateDStream = lines.flatMap(_.split(" "))
          .map((_,1))
          .updateStateByKey[Int](updateFunc)
        // stateDStream.print()
        stateDStream.foreachRDD(rdd => {
            // 内部函数
            def func(records: Iterator[(String, Int)]): Unit ={
                var conn: Connection = null
                var stmt: PreparedStatement = null
                try{
                    var url = "jdbc:mysql://localhost:3306/spark"
                    var user = "root"
                    var password = "hadoop"
                    conn = DriverManager.getConnection(url, user, password)
                    records.foreach(p=>{
                        val sql = "insert into wordcount(word,count) values (?,?)"
                        stmt = conn.prepareStatement(sql)
                        stmt.setString(1, p._1.trim)
                        stmt.setInt(2, p._2.toInt)
                        stmt.executeUpdate()
                    })
                }catch{
                    case e: Exception => e.printStackTrace()
                }finally{
                    if(stmt != null){
                        stmt.close()
                    }
                    if(conn != null){
                        conn.close()
                    }
                }
            }
            val repartitionedRDD = rdd.repartition(3)
            repartitionedRDD.foreachPartition(func)
        })
 
        ssc.start()
        ssc.awaitTermination()  // 等待终止
        spark.stop()
    }
}