一、Spark Streaming基本使用

忘是亡心i 2024-04-17 21:38 75阅读 0赞

**目录**

一、流计算

1、流计算基本介绍

2、流计算处理流程

二、Spark Streaming

1、Spark Streaming基本原理

2、Spark Streaming与Storm的对比

3、DStream操作概述

4、 基本数据源

--------------------

# 一、流计算 #

## 1、流计算基本介绍 ##

（1）、流计算秉承一个基本理念，即数据的价值随着时间的流逝而降低，

（2）、目前有三类常见的流计算框架和平台：商业级的流计算平台、开源流计算框架、公司为支持自身业务开发的流计算框架

a、商业级：IBM InfoSphere Streams和IBM StreamBase

b、较为常见的是开源流计算框架，代表如下：

*  Twitter Storm：免费、开源的分布式实时计算系统，可简单、高效、可靠地处理大量的流数据
 *  ahoo! S4（Simple Scalable Streaming System）：开源流计算平台，是通用的、分布式的、可扩展的、分区容错的、可插拔的流式系统

c、公司为支持自身业务开发的流计算框架：

*  Facebook Puma
 *  Dstream（百度）
 *  银河流数据处理平台（淘宝）

## 2、流计算处理流程 ##

**传统的数据处理流程**，需要先采集数据并存储在关系数据库等数据管理系统中，之后由用户通过查询操作和数据管理系统进行交互

![20190901173731210.png][]

**流计算的处理流程**一般包含三个阶段：**数据实时采集、数据实时计算、实时查询服务**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70][]

（1）数据实时采集

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

目前有许多互联网公司发布的开源分布式日志采集系统

*  Facebook的Scribe
 *  LinkedIn的Kafka
 *  淘宝的Time Tunnel
 *  基于Hadoop的Chukwa和Flume

# 二、Spark Streaming #

## 1、Spark Streaming基本原理 ##

Spark Streaming可整合多种输入数据源，如Kafka、Flume、HDFS，甚至是普通的TCP套接字。经处理后的数据可存储至文件系统、数据库，或显示在仪表盘里

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1][]

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

![20190901174746837.png][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2][]

## 2、Spark Streaming与Storm的对比 ##

*  Spark Streaming和Storm最大的区别在于，Spark Streaming**无法实现毫秒级的**流计算，而Storm可以实现毫秒级响应
 *  Spark Streaming构建在Spark上，一方面是因为Spark的低延迟执行引擎（100ms+）可以用于实时计算，另一方面，相比于Storm，**RDD数据集更容易做高效的容错处理**
 *  Spark Streaming采用的小批量处理的方式使得它可以**同时兼容批量和实时数据处理的逻辑和算法**，因此，方便了一些需要历史数据和实时数据联合分析的特定应用场合

## 3、DStream操作概述 ##

**（1）、Spark Streaming工作机制**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 3][]

*  在Spark Streaming中，会有一个组件 ***Receiver***，作为一个长期运行的 task 跑在***一个Executor上***
 *  每个***Receiver都会负责一个input DStream***（比如从文件中读取数据的文件流，比如套接字流，或者从Kafka中读取的一个输入流等等）
 *  Spark Streaming通过input DStream与外部数据源进行连接，读取相关数据

**（2）Spark Streaming程序的基本步骤**

*  通过创建输入DStream来**定义输入源**
 *  通过对DStream应用转**换操作和输出操作**来定义流计算
 *  用streamingContext.start()来**开始接收数据和处理流程**
 *  通过streamingContext.awaitTermination()方法来**等待处理结束**（手动结束或因为错误而结束）
 *  可以**通过 streamingContext.stop() 来手动结束流计算进程**

**（3）、创建StreamingContext对象**

import org.apache.spark._
    import org.apache.spark.streaming._
    val conf = new SparkConf().setAppName("TestDStream").setMaster("local[2]")
    val ssc = new StreamingContext(conf, Seconds(1))

## 4、 基本数据源 ##

**（1）文件流**

cd /usr/local/spark/mycode
    mkdir streaming
    cd streaming
    mkdir logfile
    cd logfile

import org.apache.spark._ 
    import org.apache.spark.streaming._
    object WordCountStreaming {  
      def main(args: Array[String]) {  
    //设置为本地运行模式，2个线程，一个监听，另一个处理数据    
        val sparkConf = new SparkConf().setAppName("WordCountStreaming").setMaster("local[2]")
        val ssc = new StreamingContext(sparkConf, Seconds(2))// 时间间隔为2秒    
    //这里采用本地文件，当然你也可以采用HDFS文件
        val lines = ssc.textFileStream("file:///usr/local/spark/mycode/streaming/logfile")  
        val words = lines.flatMap(_.split(" "))  
        val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)  
        wordCounts.print()  
        ssc.start()  
        ssc.awaitTermination()  
      }  
    }

在“/usr/local/spark/mycode/streaming/logfile”目录下新建一个log.txt文件，就可以在监听窗口中显示词频统计结果

**（2）套接字流**

**a、Socket工作原理**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 4][]

**b、Streaming代码**

NetworkWordCount

import org.apache.spark._
    import org.apache.spark.streaming._
    import org.apache.spark.storage.StorageLevel
    object NetworkWordCount {
      def main(args: Array[String]) {
        if (args.length < 2) {
          System.err.println("Usage: NetworkWordCount <hostname> <port>")
          System.exit(1)
        }
        StreamingExamples.setStreamingLogLevels()    
        val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[2]")
        val ssc = new StreamingContext(sparkConf, Seconds(1))
        val lines = ssc.socketTextStream(args(0), args(1).toInt, StorageLevel.MEMORY_AND_DISK_SER)
        val words = lines.flatMap(_.split(" "))
        val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
        wordCounts.print()
        ssc.start()
        ssc.awaitTermination()
      }
    }

StreamingExamples

package org.apache.spark.examples.streaming
    import org.apache.spark.internal.Logging
    import org.apache.log4j.{Level, Logger}
    /** Utility functions for Spark Streaming examples. */
    object StreamingExamples extends Logging {
      /** Set reasonable logging levels for streaming if the user has not configured log4j. */
      def setStreamingLogLevels() {
        val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
        if (!log4jInitialized) {
          // We first log something to initialize Spark's default logging, then we override the
          // logging level.
          logInfo("Setting log level to [WARN] for streaming example." +
            " To override add a custom log4j.properties to the classpath.")
          Logger.getRootLogger.setLevel(Level.WARN)
        }
      }
    }

**c、构建Socket数据源**

**第一种方式linux 下**

#可以在nc窗口中随意输入一些单词，监听窗口就会自动获得单词数据流信息，
    nc -lk 9999

**第二种方式：自定义数据源**

DataSourceSocket

Import  java.io.{PrintWriter}
    Import  java.net.ServerSocket
    Import  scala.io.Source
    object DataSourceSocket {
      def index(length: Int) = {
        val rdm = new java.util.Random
        rdm.nextInt(length)
      }
      def main(args: Array[String]) {
        if (args.length != 3) {
          System.err.println("Usage: <filename> <port> <millisecond>")
          System.exit(1)
        }
        val fileName = args(0)
        val lines = Source.fromFile(fileName).getLines.toList
        val rowCount = lines.length
        val listener = new ServerSocket(args(1).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(args(2).toLong)
                val content = lines(index(rowCount))
                println(content)
                out.write(content + '\n')
                out.flush()
              }
              socket.close()
            }
          }.start()
        }
      }
    }

这个窗口会不断打印出一些随机读取到的文本信息，这些信息也是Socket数据源，会被监听程序捕捉到

**（3）RDD队列流**

在**调试 Spark Streaming** 应用程序的时候，我们可以使用streamingContext.queueStream(queueOfRDD)创建基于RDD队列的DStream

新建一个 QueueStream.scala 代码文件，功能是：每隔1秒创建一个RDD，Streaming每隔2秒就对数据进行处理

import org.apache.spark.SparkConf
    import org.apache.spark.rdd.RDD
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    object QueueStream {
      def main(args: Array[String]) {
        val sparkConf = new SparkConf().setAppName("TestRDDQueue").setMaster("local[2]")
        val ssc = new StreamingContext(sparkConf, Seconds(2))
        val rddQueue =new scala.collection.mutable.SynchronizedQueue[RDD[Int]]()
        val queueStream = ssc.queueStream(rddQueue)
        val mappedStream = queueStream.map(r => (r % 10, 1))
        val reducedStream = mappedStream.reduceByKey(_ + _)
        reducedStream.print()
        ssc.start()
        for (i <- 1 to 10){
          rddQueue += ssc.sparkContext.makeRDD(1 to 100,2)
          Thread.sleep(1000)
        }
        ssc.stop()
      }
    }

[20190901173731210.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/6df3f2c48e3148a7a7f4f50c1264bf24.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/c70f77352d8047a2881b1157b034d865.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/4e5a2099818c4a1aa5560e7f61999e35.png
[20190901174746837.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/24016492b2b544f6b5dff75a070fb805.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/c2ff311eb788441e9e7489925d94da01.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/c127d0a81bd24b3f8ca76c2d9f386e35.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3MyOTQ4NzgzMDQ_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/890753ff0ed646deb4ba02e00bcf8d45.png