Spark Streaming实战之WordCount计算

一、需求1

从TCP Socket数据源实时消费数据，对每批次Batch数据进行词频统计WordCount，流程图如下：

二、准备工作

本地使用nc命令，利用它向8888端口发送数据（备注：nc是netcat的简称，原本是设置路由器），输入命令如下所示：


spark streaming实现逻辑具体代码，如下所示：

package com.ml.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}


object WordCount01 {
  def main(args: Array[String]): Unit = {
    //TODO 1、准备环境
    val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //TODO 2、加载数据
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("127.0.0.1", 8888)
    //TODO 3、处理数据
    val resultDS: DStream[(String, Int)] = lines.flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
    //TODO 4、输出数据
    resultDS.print()
    //TODO 5、启动并等待结束
    ssc.start()
    ssc.awaitTermination()
    //TODO 6、关闭资源 --优雅关闭
    ssc.stop(stopSparkContext = true, stopGracefully = true)
  }
}

运行结果，如截图所示：

三、需求2

对从Socket接收的数据做WordCount并要求能够和历史数据进行累加
如：先发一个spark，得到spark,1，然后不管隔多久再发一个spark，得到spark,2也就是说要对数据的历史状态进行维护！


代码如下所示：

package com.ml.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}


object WordCount02 {
  def main(args: Array[String]): Unit = {
    //TODO 1、准备环境
    val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //Exception in thread "main" java.lang.IllegalArgumentException: requirement failed: The checkpoint directory has not been set. Please set it by StreamingContext.checkpoint().
    ssc.checkpoint("./ckp")
    //TODO 2、加载数据
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("127.0.0.1", 8888)
    //TODO 3、处理数据
    //定义一个函数用来处理状态：把当前数据和历史状态进行累加
    //currentValues:表示该key（如：spark）的当前批次的值，如：[1,1]
    //historyValue:表示该key(如：spark)的历史值，第一次是0，后面就是之前的累加值加1
    val updateFunc = (currentValues: Seq[Int], historyValue: Option[Int]) => {
      if (currentValues.size > 0) {
        val currentResult: Int = currentValues.sum + historyValue.getOrElse(0)
        Some(currentResult)
      } else {
        historyValue
      }
    }
    val resultDS: DStream[(String, Int)] = lines.flatMap(_.split(" "))
      .map((_, 1))
      //.reduceByKey(_ + _)
      .updateStateByKey(updateFunc)
    //TODO 4、输出数据
    resultDS.print()
    //TODO 5、启动并等待结束
    ssc.start()
    ssc.awaitTermination()
    //TODO 6、关闭资源 --优雅关闭
    ssc.stop(stopSparkContext = true, stopGracefully = true)
  }
}

运行结果如下图所示：