通过案例对SparkStreaming透彻理解1

Spark在 Spark Core 之上提供了很多面向不同使用场景的高层API。比如 Spark Streaming、Spark SQL 、GraphX 、MLlib

选择spark streaming 做为源码定制的出发点的原因&＃xff1a;

1. 从依赖的专业知识上讲&＃xff0c;相对于 其他API &＃xff0c;无需引入过多的专业领域的依赖知识。

2. 从技术层面上讲&＃xff0c;是在原有Spark Core基础上 升了一维。而这是Streaming特有的。

3. 实时流处理是使用场景最广阔的&＃xff0c;是最优吸引力的。

4. 可以在Streaming处理后&＃xff0c;调用Spark兄弟框架&＃xff0c;如MLlib、SparkSQL

5. Streaming 是最复杂的&＃xff0c;因为数据一直在变动。是挑战最大的。

因此&＃xff0c;搞定Spark Streaming之后&＃xff0c;再学其他API&＃xff0c;就类似《三体》中的降维打击一样&＃xff0c;很轻松即可理解。

Streaming因为多了感知数据的逻辑&＃xff0c;因此更像是Spark上的一个应用程序。

下面实战演示&＃xff0c;实现从源源不断的输入流中过滤掉黑名单中的数据。

import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}object BlackListFilterSelfScala {def main(args: Array[String]) {val sparkConf &＃61; new SparkConf().setAppName("BlackListFilterSelfScala").setMaster("spark://master:7077")val sc &＃61; new SparkContext(sparkConf)/*** 给定默认的黑名单&＃xff0c;此数据也可以从其他数据源动态获取*/val black_list &＃61; sc.parallelize(Array("fail", "sad")).map(black_word &＃61;> (black_word, black_word))/*** 指定checkpoint*/sc.setCheckpointDir("hdfs://master:9000/library/streaming/black_list_filter/")val ssc &＃61; new StreamingContext(sc, Durations.seconds(30))/*** 输入格式&＃xff1a;关键字1,关键字2,...*/val input_word &＃61; ssc.socketTextStream("localhost", 9999)val flattenWord &＃61; input_word.flatMap(_.split(" ")).map(row &＃61;> {(row, row)})val not_black_word &＃61; flattenWord.transform(fw &＃61;> {fw.leftOuterJoin(black_list). // 左连接filter(_._2._2.isEmpty). // 将黑名单中的过滤掉map(_._1) // 只返回关键字})not_black_word.print // 输出ssc.startssc.awaitTerminationsc.stop}}

部署到集群环境中&＃xff0c;另起命令行&＃xff1a;

输出&＃xff1a;

输入的said和fail 被过滤i到了哦。成功。

成功输出后咱们马上结束程序&＃xff0c;使之只执行一次。便于之后分析。

再到 http://master:8080/ 中查看下详细内容

点击Completed Applications 的Name中 看下&＃xff1a;

我们的逻辑很简单&＃xff0c;只是创建了一个黑名单&＃xff0c;并将流进来的数据过滤掉黑名单中的关键字。

并且咱们的批处理只执行一次之后就马上结束了。不存在多次执行的情况。那为什么会有这么多的job呢&＃xff1f;

带着疑问&＃xff0c;我们看看每个job中都些什么鬼。

此处可见&＃xff0c;第一个job0是生产黑名单的job。但是代码里并没有reduceByKey。哪来的呢&＃xff1f;此处留一个悬念。

再看看Job1

原来job1是专门用来接受数据的。接收数据的Receiver也专门有一个job。这个任务运行了28S&＃xff0c;咱们任务的间隔是30S&＃xff0c;并不是28S。疑问来了。

不过至少可以知道&＃xff0c;数据是通过在Executor上运行job的方式来接收的。

Job2一看就知道是我们的业务逻辑。

这里为什么重复了呢&＃xff1f;

让我们尽请期待后续的讲解。

让我们再看看DStream&＃xff0c;DStream是离散流&＃xff1b;可以认为数据像水龙头中流出的水&＃xff0c;DStream是下面的一个水桶&＃xff0c;在设定的时间间隔内换一个空桶&＃xff0c;之前的水桶取走&＃xff0c;交给下游流水线处理(业务处理)。

此处可以看到&＃xff0c;只是按设置的时间来触发换水桶&＃xff0c;换句话说&＃xff0c;只受时间维度的影响&＃xff0c;因为&＃xff0c;时间对所有的业务理解都是一样的&＃xff0c;此设计是超棒的解耦。