Spark Streaming流式计算的WordCount入门

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

内部处理流程：

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

Java代码

1. //下面不需要使用的依赖，大家可根据情况去舍
2. name := "scala-spark"
4. version := "1.0"
6. scalaVersion := "2.11.7"
8. //使用公司的私服
9. resolvers += "Local Maven Repository" at "http://dev.bizbook-inc.com:8083/nexus/content/groups/public/"
11. //使用内部仓储
12. externalResolvers := Resolver.withDefaultResolvers(resolvers.value, mavenCentral = false)
14. //Hadoop的依赖
15. libraryDependencies += "org.apache.hadoop" % "hadoop-client" % "2.7.1" //% "provided"
17. //Habse的依赖
18. libraryDependencies += "org.apache.hbase" % "hbase-client" % "0.98.12-hadoop2" // % "provided"
20. libraryDependencies += "org.apache.hbase" % "hbase-common" % "0.98.12-hadoop2" //% "provided"
22. libraryDependencies += "org.apache.hbase" % "hbase-server" % "0.98.12-hadoop2" //% "provided"
24. //Spark的依赖
25. libraryDependencies += "org.apache.spark" % "spark-core_2.11" % "1.6.0" //% "provided"
27. //Spark SQL 依赖
28. libraryDependencies += "org.apache.spark" % "spark-sql_2.11" % "1.6.0" //% "provided"
30. //Spark For Hive 依赖
31. libraryDependencies += "org.apache.spark" % "spark-hive_2.11" % "1.6.0"
34. //Spark for Streaming
35. libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % "1.6.0"
38. //java servlet 依赖
39. libraryDependencies += "javax.servlet" % "javax.servlet-api" % "3.0.1" //% "provided"

Java代码

1. package com.tools.streaming
3. import org.apache.spark.SparkConf
4. import org.apache.spark.streaming._
6. /**
7. * Created by qindongliang on 2016/1/28.
8. */
9. object StreamingWordCount {
12. def main(args: Array[String]) {
13. //开本地线程两个处理
14. val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
15. //每隔10秒计算一批数据
16. val ssc = new StreamingContext(conf, Seconds(10))
17. //监控机器ip为192.168.1.187:9999端号的数据,注意必须是这个9999端号服务先启动nc -l 9999，否则会报错,但进程不会中断
18. val lines = ssc.socketTextStream("192.168.1.187", 9999)
19. //按空格切分输入数据
20. val words = lines.flatMap(_.split(" "))
21. //计算wordcount
22. val pairs = words.map(word => (word, 1))
23. //word ++
24. val wordCounts = pairs.reduceByKey(_ + _)
25. //排序结果集打印，先转成rdd，然后排序true升序，false降序，可以指定key和value排序_._1是key，_._2是value
26. val sortResult=wordCounts.transform(rdd=>rdd.sortBy(_._2,false))
27. sortResult.print()
28. ssc.start() // 开启计算
29. ssc.awaitTermination() // 阻塞等待计算
31. }
34. }

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

Java代码

1. nc -l 9999
2. 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