美文网首页Java高级进阶
Spark Streaming--实战篇

Spark Streaming--实战篇

作者: Java大生 | 来源:发表于2018-09-25 21:31 被阅读2次

    摘要:

       Sprak Streaming属于Saprk API的扩展,支持实时数据流(live data streams)的可扩展,高吞吐(hight-throughput) 容错(fault-tolerant)的流处理。可以接受来自KafKa,Flume,ZeroMQ Kinesis Twitter或TCP套接字的数据源,处理的结果数据可以存储到文件系统 数据库 现场dashboards等。

    DStream编程模型

    Dstream是Spark streaming中的高级抽象连续数据流,这个数据源可以从外部获得(如KafKa Flume等),也可以通过输入流获得,还可以通过在其他DStream上进行高级操作创建,DStream是通过一组时间序列上连续的RDD表示的,所以一个DStream可以看作是一个RDDs的序列。

    DStream操作

    1.套接字流:通过监听Socket端口来接收数据。

    通过Scala编写程序来产生一系列的字符作为输入流:

    GenerateChar:

    object GenerateChar {

      def generateContext(index : Int) : String = {

        import scala.collection.mutable.ListBuffer

        val charList = ListBuffer[Char]()

        for(i <- 65 to 90)

          charList += i.toChar

        val charArray = charList.toArray

        charArray(index).toString

      }

      def index = {

        import  java.util.Random

        val rdm = new Random

        rdm.nextInt(7)

      }

      def main(args: Array[String]) {

        val listener = new ServerSocket(9998)

        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(500)

                val context = generateContext(index)  //产生的字符是字母表的前七个随机字母

                println(context)

                out.write(context + '\n')

                out.flush()

              }

              socket.close()

            }

          }.start()

        }

      }

    }

    ScoketStreaming:

    object ScoketStreaming {

      def main(args: Array[String]) {

        //创建一个本地的StreamingContext,含2个工作线程

        val conf = new SparkConf().setMaster("local[2]").setAppName("ScoketStreaming")

        val sc = new StreamingContext(conf,Seconds(10))  //每隔10秒统计一次字符总数

        //创建珍一个DStream,连接master:9998

        val lines = sc.socketTextStream("master",9998)

        val words = lines.flatMap(_.split(" "))

        val wordCounts = words.map(x => (x , 1)).reduceByKey(_ + _)

        wordCounts.print()

        sc.start()        //开始计算

        sc.awaitTermination()  //通过手动终止计算,否则一直运行下去

      }

    }

    运行结果:

    GenerateChar产生的数据如下:

    Got client connected from :/192.168.31.128

    ScoketStreaming运行结果:

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

    Time: 1459426750000 ms

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

    (B,1)

    (G,1)

    (C,1)

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

    Time: 1459426760000 ms

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

    (B,5)

    (F,3)

    (D,4)

    (G,3)

    (C,3)

    (E,1)

    注意:如果是在本地运行的,setMaster的参数必须为local[n],n >1,官网解释:

      When running a Spark Streaming program locally, do not use “local” or “local[1]” as the master URL. Either ofthese means that only one thread

    will be used for running tasks locally. If you are using a input DStream based on a receiver (e.g. sockets, Kafka, Flume, etc.), then the single

    thread will be used to run the receiver,leaving no thread for processing the received data.

      当在本地运行Spark Streaming程序时,Master的URL不能设置为"local"或"local[1]",这两种设置都意味着你将会只有一个线程来运行作业,如果你的Input DStream基于一个接收器

    (如Kafka,Flume等),那么只有一个线程来接收数据,而没有多余的线程来处理接收到的数据。

    如果是在集群上运行,为Spark streaming应分配的核数应该在大于接收器的数据,否则同样只接收了数据而没有能力处理。

    2.文件流:Spark Streaming通过监控文件系统的变化,若有新文件添加,则将它读入并作为数据流

    需要注意的是:

     1.这些文件具有相同的格式

     2.这些文件通过原子移动或重命名文件的方式在dataDirectory创建

     3.一旦移动这些文件,就不能再进行修改,如果在文件中追加内容,这些追加的新数据也不会被读取。

    FileStreaming:

    object FileStreaming {

      def main(args: Array[String]) {

        val conf = new SparkConf().setMaster("local").setAppName("FileStreaming")

        val sc = new StreamingContext(conf,Seconds(5))

        val lines = sc.textFileStream("/home/hadoop/wordCount")

        val words = lines.flatMap(_.split(" "))

        val wordCounts = words.map(x => (x , 1)).reduceByKey(_ + _)

        sc.start()

        sc.awaitTermination()

      }

    }

    当你在文件目录里添加文件时,Spark Streaming就会自动帮你读入并计算 ,可以读取本地目录 HDFS和其他文件系统。

    注意:文件流不需要运行接收器,所以不需要分配核数

    3.RDD队列流:使用streamingContext.queueStream(queueOfRDD)创建基于RDD队列的DStream,用于调试Spark Streaming应用程序。

    QueueStream:程序每隔1秒就创建一个RDD,Streaming每隔1秒就就对数据进行处理

    object QueueStream {

      def main(args: Array[String]) {

        val conf = new SparkConf().setMaster("local[2]").setAppName("queueStream")

        //每1秒对数据进行处理

        val ssc = new StreamingContext(conf,Seconds(1))

        //创建一个能够push到QueueInputDStream的RDDs队列

        val rddQueue = new mutable.SynchronizedQueue[RDD[Int]]()

        //基于一个RDD队列创建一个输入源

        val inputStream = ssc.queueStream(rddQueue)

        val mappedStream = inputStream.map(x => (x % 10,1))

        val reduceStream = mappedStream.reduceByKey(_ + _)

        reduceStream.print

        ssc.start()

        for(i <- 1 to 30){

          rddQueue += ssc.sparkContext.makeRDD(1 to 100, 2)  //创建RDD,并分配两个核数

          Thread.sleep(1000)                                 

        }

        ssc.stop()

      }

    }

    输出

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

    Time: 1459595433000 ms //第1个输出

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

    (4,10)

    (0,10)

    (6,10)

    (8,10)

    (2,10)

    (1,10)

    (3,10)

    (7,10)

    (9,10)

    (5,10)

    ............

    ............

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

    Time: 1459595463000 ms //第30个输出

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

    (4,10)

    (0,10)

    (6,10)

    (8,10)

    (2,10)

    (1,10)

    (3,10)

    (7,10)

    (9,10)

    (5,10) 

    4.带状态的处理staefull

    updateStateByKey操作:使用updateStateByKey操作的地是为了保留key的状态,并能持续的更新;使用此功能有如下两个步骤:

     1.定义状态,这个状态可以是任意的数据类型

     2.定义状态更新函数, 指定一个函数根据之前的状态来确定如何更新状态。

    同样以wordCount作为例子,不同的是每一次的输出都会累计之前的wordCount

    StateFull:

    object StateFull {

      def main(args: Array[String]) {

        //定义状态更新函数

        val updateFunc = (values: Seq[Int], state: Option[Int]) => {

          val currentCount = values.foldLeft(0)(_ + _)

          val previousCount = state.getOrElse(0)

          Some(currentCount + previousCount)

        }

        val conf = new SparkConf().setMaster("local[2]").setAppName("stateFull")

        val sc = new StreamingContext(conf, Seconds(5))

        sc.checkpoint(".")    //设置检查点,存储位置是当前目录,检查点具有容错机制

        val lines = sc.socketTextStream("master", 9999)

        val words = lines.flatMap(_.split(" "))

        val wordDstream = words.map(x => (x, 1))

        val stateDstream = wordDstream.updateStateByKey[Int](updateFunc)

        stateDstream.print()

        sc.start()

        sc.awaitTermination()

      }

    }

    先运行之前GenerateChar来产生字母,再运行StateFull,结果如下:

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

    Time: 1459597690000 ms

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

    (B,3)

    (F,1)

    (D,1)

    (G,1)

    (C,1)

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

    Time: 1459597700000 ms //会累计之前的值

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

    (B,5)

    (F,3)

    (D,4)

    (G,4)

    (A,2)

    (E,5)

    (C,4)

    Spark Straming最大的优点在于处理数据采用的是粗粒度的处理方式(一次处理一小批的数据),这种特性也更方便地实现容错恢复机制,其DStream是在RDD上的高级

    抽象,所以其极容易与RDD进行互操作

    相关文章

      网友评论

        本文标题:Spark Streaming--实战篇

        本文链接:https://www.haomeiwen.com/subject/pckroftx.html