park Streaming--实战篇

摘要：

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

C

G

B

C

F

G

D

G

B

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进行互操作。

巴特西

2.spark-streaming实战

park Streaming--实战篇

最新文章

热门文章