Spark Streaming小结 | 程序员的自我修养 | 关注Java、大数据、机器学习

概述

Spark Streaming类似于Apache Storm，用于流式数据的处理。根据其官方文档介绍，Spark Streaming有高吞吐量和容错能力强这两个特点。Spark Streaming支持的数据输入源很多，例如：Kafka、Flume、Twitter、ZeroMQ和简单的TCP套接字等等。数据输入后可以用Spark的高度抽象原语如：map、reduce、join、window等进行运算。而结果也能保存在很多地方，如HDFS，数据库等。另外Spark Streaming也能和MLlib（机器学习）以及Graphx完美融合。

其内部工作方式如下：

Word Count示例

private static final Pattern SPACE = Pattern.compile(" ");
public static void main(String[] args) {
	StreamingExamples.setStreamingLogLevels();

	JavaStreamingContext jssc = new JavaStreamingContext("local[2]",
		  "JavaNetworkWordCount", new Duration(10000));
	jssc.checkpoint(".");//使用updateStateByKey()函数需要设置checkpoint
	//打开本地的端口9999
	JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);
	//按行输入，以空格分隔
	JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(SPACE.split(line)));
	//每个单词形成pair，如（word，1）
	JavaPairDStream<String, Integer> pairs = words.mapToPair(word -> new Tuple2<>(word, 1));
	//统计并更新每个单词的历史出现次数
	JavaPairDStream<String, Integer> counts = pairs.updateStateByKey((values, state) -> {
		Integer newSum = state.or(0);
		for(Integer i :values) {
			newSum += i;
		}
		return Optional.of(newSum);
	});
	counts.print();
	jssc.start();
	jssc.awaitTermination();
}

private static final Pattern SPACE = Pattern.compile(" ");

public static void main(String[] args) {

StreamingExamples.setStreamingLogLevels();

JavaStreamingContext jssc = new JavaStreamingContext("local[2]",

"JavaNetworkWordCount", new Duration(10000));

jssc.checkpoint(".");//使用updateStateByKey()函数需要设置checkpoint

//打开本地的端口9999

JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);

//按行输入，以空格分隔

JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(SPACE.split(line)));

//每个单词形成pair，如（word，1）

JavaPairDStream<String, Integer> pairs = words.mapToPair(word -> new Tuple2<>(word, 1));

//统计并更新每个单词的历史出现次数

JavaPairDStream<String, Integer> counts = pairs.updateStateByKey((values, state) -> {

Integer newSum = state.or(0);

for(Integer i :values) {

newSum += i;

}

return Optional.of(newSum);

});

counts.print();

jssc.start();

jssc.awaitTermination();

}

启动Netcat

$ nc -lk 9999

1	$ nc -lk 9999

启动Spark Streaming Application

若在本地调试，可在IDE中启动，否则，用如下命令启动：

$ ./bin/run-example org.apache.spark.examples.streaming.JavaNetworkWordCount localhost 9999

1	$ ./bin/run-example org.apache.spark.examples.streaming.JavaNetworkWordCount localhost 9999

测试

输入：

hello world
hello spark
hello yurnom

hello world

hello spark

hello yurnom

结果：

-------------------------------------------
Time: 1407741020000 ms
-------------------------------------------
(yurnom,1)
(hello,3)
(world,1)
(spark,1)

-------------------------------------------

Time: 1407741020000 ms

-------------------------------------------

(yurnom,1)

(hello,3)

(world,1)

(spark,1)

DStream

Discretized Stream是Spark Streaming的基础抽象，代表持续性的数据流和经过各种Spark原语操作后的结果数据流。在内部实现上，DStream由连续的序列化RDD来表示。每个RDD含有一段时间间隔内的数据，如下图：

对数据的操作也是按照RDD为单位来进行的，如下图所示：

上图下方的RDD都是通过Spark高级原语的转换而来，计算过程由Spark engine来完成。

Operations

DStream上的原语与RDD的类似，分为Transformations（转换）和Output Operations（输出）两种，此外转换操作中还有一些比较特殊的原语，如：updateStateByKey()、transform()以及各种Window相关的原语。

UpdateStateByKey Operation

UpdateStateByKey原语用于记录历史记录，上文中Word Count示例中就用到了该特性。若不用UpdateStateByKey来更新状态，那么每次数据进来后分析完成后，结果输出后将不在保存。如，若将上文代码示例中的第15行若替换为：

JavaPairDStream<String, Integer> counts = pairs.reduceByKey((i1, i2) -> (i1 + i2));

1	JavaPairDStream<String, Integer> counts = pairs.reduceByKey((i1, i2) -> (i1 + i2));

那么输入：hellow world，结果则为：(hello,1)(world,1)，然后输入hello spark，结果则为(hello,1)(spark,1)。也就是不会保留上一次数据处理的结果。

使用UpdateStateByKey原语需要用于记录的State，可以为任意类型，如上例中即为Optional<Intege>类型；此外还需要更新State的函数，可参考Word Count示例中的15-20行。

Transform Operations

Transform()原语允许DStream上执行任意的RDD-to-RDD函数。通过该函数可以方便的扩展Spark API。此外，本篇开头所提到的MLlib（机器学习）以及Graphx也是通过本函数来进行结合的。官方示例：

import org.apache.spark.streaming.api.java.*;
// RDD containing spam information
final JavaPairRDD<String, Double> spamInfoRDD = jssc.sparkContext().newAPIHadoopRDD(...);

JavaPairDStream<String, Integer> cleanedDStream = wordCounts.transform(
  new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
    @Override public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> rdd) throws Exception {
      rdd.join(spamInfoRDD).filter(...); // join data stream with spam information to do data cleaning
      ...
    }
  });

import org.apache.spark.streaming.api.java.*;

// RDD containing spam information

final JavaPairRDD<String, Double> spamInfoRDD = jssc.sparkContext().newAPIHadoopRDD(...);

JavaPairDStream<String, Integer> cleanedDStream = wordCounts.transform(

new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {

@Override public JavaPairRDD<String, Integer> call(JavaPairRDD<String, Integer> rdd) throws Exception {

rdd.join(spamInfoRDD).filter(...); // join data stream with spam information to do data cleaning

...

}

});

Window Operations

Window Operations有点类似于Storm中的State，可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Steaming的允许状态。如下图所示：

如以下代码表示，每10秒钟处理最近30秒钟中的数据。

JavaPairDStream<String, Integer> windowedWordCounts = 
	  pairs.reduceByKeyAndWindow((a, b) -> (a + b), 
			new Duration(30000), new Duration(10000));

JavaPairDStream<String, Integer> windowedWordCounts =

pairs.reduceByKeyAndWindow((a, b) -> (a + b),

new Duration(30000), new Duration(10000));

Window相关API有：

window(windowLength, slideInterval)
countByWindow(windowLength, slideInterval)
reduceByWindow(func, windowLength, slideInterval)
reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])
reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])
countByValueAndWindow(windowLength, slideInterval, [numTasks])

Output Operations

当某个Output Operations原语被调用时，stream才会开始真正的计算过程。现阶段支持的Output方式有以下几种：

print()
foreachRDD(func)
saveAsObjectFiles(prefix, [suffix])
saveAsTextFiles(prefix, [suffix])
saveAsHadoopFiles(prefix, [suffix])

其它特性

输入源

除了前文中Word Count示例中用到的TCP套接字连接连接作为输入源以外，Spark Streaming还可以使用很多其它的输入源。例如对于文件，可以这样处理：

jssc.fileStream(dataDirectory);

1	jssc.fileStream(dataDirectory);

Spark Streaming将会监控该文件夹，要使用该特性，需要注意以下几点：

该文件夹下的所有文件必须有相同的数据格式
在该文件夹下创建文件的方式必须是原子性的移动或重命名的方式，不可以先创建文件后在进行写入
所有文件夹下的文件不可进行改动

其它数据源的使用可以参考Spark安装包中的examples文件夹中的streaming部分。同样对于特殊的数据输入源，可以进行定制。

监控

一般来说，使用Spark自带的Web UI就能满足大部分的监控需求。对于Spark Streaming来说，以下两个度量指标尤为重要（在Batch Processing Statistics标签下）：

Processing Time：处理每个batch的时间
Scheduling Delay：每个batch在队列中等待前一个batch完成处理所等待的时间

若Processing Time的值一直大于Scheduling Delay，或者Scheduling Delay的值持续增长，代表系统已经无法处理这样大的数据输入量了，这时就需要考虑各种优化方法来增强系统的负载。

持久化

与RDD一样，DStream同样也能通过persist()方法将数据流存放在内存中，这样做的好处是遇到需要多次迭代计算的程序时，速度优势十分的明显。而对于上文中提到的各种window原语，其默认的持久化策略就是保存在内存中。

当数据源来自于网络时（例如通过Kafka、Flume、sockets等等），默认的持久化策略是将数据保存在两台机器上，这也是为了容错性而设计的。

参考资料

（转载本站文章请注明作者和出处程序员的自我修养 – SelfUp.cn ，请勿用于任何商业用途）

ascend: Error:(36, 29) wrong number of type parameters for overloaded method value createDirectStream with...
Anonymous:
Anonymous:
Anonymous:
Anonymous:
Anonymous: 运行报错：java.lang.Abstr actMethodError
Anonymous:
Anonymous:
Anonymous:
Anonymous:
zzg: 请问zz你问题核实了没� ��我也遇到了这个奇葩� ��问题，我们也是基�...
Anonymous:
kuyuzasur: what do i tell doctor to get viagra. edhelprie.mdhelpserv.com – how long we can take viagra,
Anonymous: 这个问题可以忽略。不影响
Anonymous:
Anonymous: 牛
Anonymous: 楼主你好，我偶尔也会遇到Reconnect due to socket error: java.nio.channels.ClosedCha...
Anonymous: sdfs
Anonymous:
Anonymous: java.io.NotSerializableExcepti on: DStream checkpointing has been enabled but the DStreams with their...

9条评论

匿名说道：

2015年4月28日下午8:43

回复
aaa说道：

2015年5月27日下午9:33

这个例子也就只能基于nc的测试而已~~~·

回复
匿名说道：

2015年9月9日上午9:43

回复
匿名说道：

2015年10月8日下午3:04

回复
匿名说道：

2016年7月1日下午3:08

回复
匿名说道：

2016年8月4日上午10:16

楼主，请问这句代码哪里来：
StreamingExamples.setStreamingLogLevels();

？

请指教

回复
匿名说道：

2017年2月9日下午5:18

17/02/09 01:15:02 WARN Utils: Service 'SparkUI' could not bind on port 4040. Attempting port 4041.
卡在这一行不动了
给个高招吧

回复
- 一波清泉说道：
  
  2017年2月9日下午5:19
  
  回复邮箱： 1004161699@qq.com
  多谢
  
  回复
匿名说道：

2017年7月14日下午2:56

回复

发表评论

点击这里取消回复。