Spark Streaming流式计算的WordCount入门

Spark Streaming是一种近实时的流式计算模型，它将作业分解成一批一批的短小的批处理任务，然后并行计算，具有可扩展，高容错，高吞吐，实时性高等一系列优点，在某些场景可达到与Storm一样的处理程度或优于storm，也可以无缝集成多重日志收集工具或队列中转器，比如常见的 kakfa，flume，redis，logstash等，计算完后的数据结果，也可以
存储到各种存储系统中，如HDFS，数据库等，一张简单的数据流图如下：



内部处理流程：








下面来看一个wordcount级别的入门例子,注意需要导入相关的包：

//下面不需要使用的依赖，大家可根据情况去舍
name := "scala-spark"

version := "1.0"

scalaVersion := "2.11.7"

//使用公司的私服
resolvers += "Local Maven Repository" at "http://dev.bizbook-inc.com:8083/nexus/content/groups/public/"

//使用内部仓储
externalResolvers := Resolver.withDefaultResolvers(resolvers.value, mavenCentral = false)

//Hadoop的依赖
libraryDependencies += "org.apache.hadoop" % "hadoop-client" % "2.7.1" //% "provided"

//Habse的依赖
libraryDependencies += "org.apache.hbase" % "hbase-client" % "0.98.12-hadoop2" // % "provided"

libraryDependencies += "org.apache.hbase" % "hbase-common" % "0.98.12-hadoop2"  //% "provided"

libraryDependencies += "org.apache.hbase" % "hbase-server" % "0.98.12-hadoop2" //% "provided"

//Spark的依赖
libraryDependencies += "org.apache.spark" % "spark-core_2.11" % "1.6.0" //% "provided"

//Spark SQL 依赖
libraryDependencies += "org.apache.spark" % "spark-sql_2.11" % "1.6.0" //% "provided"

//Spark For Hive 依赖
libraryDependencies += "org.apache.spark" % "spark-hive_2.11" % "1.6.0"


//Spark for Streaming
libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % "1.6.0"


//java servlet 依赖
libraryDependencies += "javax.servlet" % "javax.servlet-api" % "3.0.1" //% "provided"

[/size]

package com.tools.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming._

/**
  * Created by qindongliang on 2016/1/28.
  */
object StreamingWordCount {


  def main(args: Array[String]) {
    //开本地线程两个处理
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    //每隔10秒计算一批数据
    val ssc = new StreamingContext(conf, Seconds(10))
    //监控机器ip为192.168.1.187:9999端号的数据,注意必须是这个9999端号服务先启动nc -l 9999，否则会报错,但进程不会中断
    val lines = ssc.socketTextStream("192.168.1.187", 9999)
    //按空格切分输入数据
    val words = lines.flatMap(_.split(" "))
    //计算wordcount
    val pairs = words.map(word => (word, 1))
    //word ++
    val wordCounts = pairs.reduceByKey(_ + _)
    //排序结果集打印，先转成rdd，然后排序true升序，false降序，可以指定key和value排序_._1是key，_._2是value
    val sortResult=wordCounts.transform(rdd=>rdd.sortBy(_._2,false))
    sortResult.print()
    ssc.start()             // 开启计算
    ssc.awaitTermination()  // 阻塞等待计算

  }


}

然后在对应的linux机器上，开一个nc服务，并写入一些数据：

nc -l 9999
a a a c c d d v v e p x x x x  o

然后在控制台，可见计算结果，并且是排好序的：






至此，第一个体验流式计算的demo就入门了，后面我们还可以继续完善这个例子，比如从kakfa或者redis里面接受数据，然后存储到hbase，或者mysql或者solr，lucene，elasticsearch索引中，用来给前端js图表绘图所用。


参考文章：

http://blog.scottlogic.com/2013/07/29/spark-stream-analysis.html
http://spark.apache.org/docs/latest/streaming-programming-guide.html


有什么问题 可以扫码关注微信公众号：我是攻城师(woshigcs)，在后台留言咨询。
本公众号的内容是有关搜索和大数据技术和互联网等方面内容的分享，也是一个温馨的技术互动交流的小家园