JavaSpark-Spark Streaming

实验使用nc作为stream连接，连接命令：C:\Users\YZ-0212\Desktop\nc.exe -l -v -p 9999

socket监听端口，是所有网络通讯的基础

许多应用需要即时处理收到的数据

例如用来实时追踪页面访问统计的应用、训练机器学 习模型的应用，还有自动检测异常的应用

Spark Streaming 是 Spark 为这些应用而设计的 模型。它允许用户使用一套和批处理非常接近的 API 来编写流式计算应用，这样就可以大 量重用批处理应用的技术甚至代码

和Spark 基于RDD 的概念很相似，Spark Streaming 使用离散化流（discretized stream）作 为抽象表示，叫作 DStream

DStream 是随时间推移而收到的数据的序列。在内部，每个时间区间收到的数据都作为 RDD 存在，而 DStream 是由这些 RDD 所组成的序列（因此得名“离散化”）

DStream可以从各种输入源创建，比如 Flume、Kafka 或者 HDFS。

创建出来的 DStream 支持两种操作，一种是转化操作（transformation），会生成一个新的 DStream，另一种是输出操作（output operation），可以把数据写入外部系统中

DStream 提供了许多与 RDD 所支持的操作相类似的操作支持，还增加了与时间相关的新操作，比如滑动窗口

Spark Streaming 应用需要进行额外配置来保证 24/7 不间断工作

检查点（checkpointing）机制，也就是把数据存储到可靠文件系统（比如 HDFS）上的机制

这也是 Spark Streaming 用来实现不间断工作的主要方式

此外，在遇到失败时如何重启应用，以及如何把应用设置为自动重启模式

例子

例子。从一台 服务器的 7777 端口上收到一个以换行符分隔的多行文本，要从中筛选出包含单词 error 的 行，并打印出来

Spark Streaming 程序最好以使用 Maven 或者 sbt 编译出来的独立应用的形式运行。Spark Streaming 虽然是 Spark 的一部分，它在 Maven 中也以独立工件的形式提供，你也需要在 工程中添加一些额外的 import 声明

们从创建 StreamingContext 开始，它是流计算功能的主要入口。StreamingContext 会 在底层创建出 SparkContext，用来处理数据

其构造函数还接收用来指定多长时间处理 一次新数据的批次间隔（batch interval）作为输入

调 用 socketTextStream() 来创建出基于本地 7777 端口上收到的文本数据的 DStream。然后 把 DStream 通过 filter() 进行转化，只得到包含“error”的行

JavaStreamingContext jssc = new JavaStreamingContext(conf, new Duration(1000));
JavaDStream<String> lines = jssc.socketTextStream("localhost", 7777);
JavaDStream<String> errorLines = lines.filter(x -> {return x.contains("erroe");});
errorLines.print();
jssc.start();
jssc.awaitTermination();

这只是设定好了要进行的计算，系统收到数据时计算就会开始。要开始接收数据，必须 显式调用 StreamingContext 的 start() 方法

执行会在另一个线程中进行，所以需要调用 awaitTermination 来等待流计算完成，来防止应用退出

，一个 Streaming context 只能启动一次，所以只有在配置好所有 DStream 以及所需 要的输出操作之后才能启动

架构与抽象

Spark Streaming 使用“微批次”的架构，把流式计算当作一系列连续的小规模批处理来对

待

Spark Streaming 从各种输入源中读取数据，并把数据分组为小的批次

新的批次按均匀的时间间隔创建出来。 在每个时间区间开始的时候，一个新的批次就创建出来，在该区间内收到的数据都会被添加到这个批次中

在时间区间结束时，批次停止增长

时间区间的大小是由批次间隔这个参数决定的，每个输入批次都形成一个 RDD， 以 Spark 作业的方式处理并生成其他的 RDD。处理的结果可以以批处理的方式传给外部系统



Spark Streaming 的编程抽象是离散化流，也就是 DStream。 它是一个 RDD 序列，每个 RDD 代表数据流中一个时间片内的数据



可以从外部输入源创建 DStream，也可以对其他 DStream 应用进行转化操作得到新的DStream，还支持许多的RDD操作

DStream 还有“有状态”的转化操作，可以用来聚合不同时间片内的数据

从套接字中接收数据，创建DStream对其应用filter() 转化操作。这会在内部创建出如下图 所示的 RDD



这里的时间间隔是创建 StreamingContext 时设定的

DStream 还支持输出操作，比如在示例中使用的 print()

输出操作和 RDD 的行动操作的概念类似

Spark 在行动操作中将数据写入外部系统中，而 SparkStreaming 的输出操作在每个时间区间中周期性执行，每个批次都生成输出

Spark Streaming 在 Spark 的驱动器程序—工作节点的结构的执行过程（对 Spark 组成部分的描述）



Spark Streaming 为每个输入源启动对应的接收器。接收器以任务的形式运行在应用的执行器进程中，从输入源收集数据并保存为 RDD。它们收集到输入数据后会把数据复制到另一个执行器进程来保障容错性（默认行为）。数据保存在执行器进程的内存中，和缓存 RDD 的方式一样。驱动器程序中的StreamingContext 会周期性地运行 Spark 作业来处理这些数据，把数据与之前时间区间中的RDD 进行整合

只要输入数据还在，它就可以使用 RDD 谱系重算出任意状态

谱系图来恢复的话，重算有可能会花很长时间，因为需要处理从程序启动以来的所有数据

检查点机制，可以把状态阶段性地存储到可靠文件系统（HDFS）中一般来说，你需要每处理 5-10 个批次的数据就保存一次。在恢复数据时，Spark Streaming 只需要回溯到上一个检查点即可

转化操作

无状态转化操作中，每个批次的处理不依赖于之前批次的数据。常见的RDD转化操作，例如map()、filter()、reduceByKey()等，都是无状态转化操作

有状态转化操作需要使用之前批次的数据或者是中间结果来计算当前批次的数

据。有状态转化操作包括基于滑动窗口的转化操作和追踪状态变化的转化操作

无状态转化操作

将 RDD 转化操作应用到每个批次上，也就是转化 DStream中的每一个 RDD

RDD 转化操作有不少都可用于DStream（针对键值对的 DStream 转化操作（比如 reduceByKey()）要添加 importStreamingContext._ 才能在 Scala 中使用）

函数名称 目　　的 Scala示例
用来操作DStream[T]
的用户自定义函数的
函数签名
map() 对 DStream 中的每个元素应用给
定函数，返回由各元素输出的元
素组成的 DStream。
ds.map(x => x + 1) f: (T) -> U
flatMap() 对 DStream 中的每个元素应用给
定函数，返回由各元素输出的迭
代器组成的 DStream。
ds.flatMap(x => x.split(" ")) f: T -> Iterable[U]
filter() 返回由给定 DStream 中通过筛选
的元素组成的 DStream。
ds.filter(x =>
12a6e
x != 1) f: T -> Boolean
repartition() 改变 DStream 的分区数。 ds.repartition(10) N/A
reduceByKey() 将每个批次中键相同的记录归约。 ds.reduceByKey(
(x, y) => x + y)
f: T, T -> T
groupByKey() 将每个批次中的记录根据键分组。 ds.groupByKey() N/A

函数看起来像作用在整个流上，但事实上每个 DStream 在内部是由许多 RDD（批次）组成，且无状态转化操作是分别应用到每个 RDD 上的

//java使用ip进行计数  map reducebykey （未实际练习）
// 假设ApacheAccessingLog是用来从Apache日志中解析条目的工具类
static final class IpTuple implements PairFunction<ApacheAccessLog, String, Long> {
public Tuple2<String, Long> call(ApacheAccessLog log) {
return new Tuple2<>(log.getIpAddress(), 1L);
}
}
JavaDStream<ApacheAccessLog> accessLogsDStream =
logData.map(new ParseFromLogLine());
JavaPairDStream<String, Long> ipDStream =
accessLogsDStream.mapToPair(new IpTuple());
JavaPairDStream<String, Long> ipCountsDStream =
ipDStream.reduceByKey(new LongSumReducer());

无状态转化操作也能在多个 DStream 间整合数据，不过也是在各个时间区间内

连接转化操作， cogroup()、join()、leftOuterJoin() 等

JavaPairDStream<String, Long> ipBytesDStream =
accessLogsDStream.mapToPair(new IpContentTuple());
JavaPairDStream<String, Long> ipBytesSumDStream =
ipBytesDStream.reduceByKey(new LongSumReducer());
JavaPairDStream<String, Tuple2<Long, Long>> ipBytesRequestCountDStream =
ipCountsDStream.join(ipBytesSumDStream);

transform() 高级操作符，可以让你直接操作其内部的 RDD

允许你对 DStream 提供任意一个 RDD 到 RDD 的函数。这个函数会在数据流中的每个批次中被调用，生成一个新的流

//是重用你为 RDD 写的批处理代码
JavaPairDStream<String, Long> ipRawDStream = accessLogsDStream.transform(
new Function<JavaRDD<ApacheAccessLog>, JavaRDD<ApacheAccessLog>>() {
public JavaPairRDD<ApacheAccessLog> call(JavaRDD<ApacheAccessLog> rdd) {
return extractOutliers(rdd);
}

通过 StreamingContext.transform 或 DStream.transformWith(otherStream, func)来整合与转化多个 DStream

有状态转化操作

DStream 的有状态转化操作是跨时间区间跟踪数据的操作；也就是说，一些先前批次的数据也被用来在新的批次中计算结果

两种类型是滑动窗口和 updateStateByKey()，前者以一个时间阶段为滑动窗口进行操作，后者则用来跟踪每个键的状态变化

有状态转化操作需要在你的 StreamingContext 中打开检查点机制来确保容错性

设置检查点
ssc.checkpoint("hdfs://...")
//进行本地开发时，你也可以使用本地路径（例如 /tmp）取代 HDFS

基于窗口的转化操作

基于窗口的操作会在一个比 StreamingContext 的批次间隔更长的时间范围内，通过整合多个批次的结果，计算出整个窗口的结果

所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长，两者都必须是

StreamContext 的批次间隔的整数倍

窗口时长控制每次计算最近的多少个批次的数据，其实就是最近的windowDuration/batchInterval 个批次

而滑动步长的默认值与批次间隔相等，用来控制对新的 DStream 进行计算的间隔。如果源 DStream 批次间隔为 10 秒，并且我们只希望每两个批次计算一次窗口结果，就应该把滑动步长设置为 20 秒

对 DStream 可以用的最简单窗口操作是 window()，它返回一个新的 DStream 来表示所请求的窗口操作的结果数据

换句话说，window() 生成的 DStream 中的每个 RDD 会包含多个批次中的数据，可以对这些数据进行 count()、transform() 等操作

//在 Java 中使用 window() 对窗口进行计数
JavaDStream<ApacheAccessLog> accessLogsWindow = accessLogsDStream.window(
Durations.seconds(30), Durations.seconds(10));
JavaDStream<Integer> windowCounts = accessLogsWindow.count();

尽管可以使用 window() 写出所有的窗口操作，Spark Streaming 还是提供了一些其他的窗口操作，让用户可以高效而方便地使用。首先，reduceByWindow() 和reduceByKeyAndWindow()让我们可以对每个窗口更高效地进行归约操作。它们接收一个归约函数，在整个窗口上执行

这两个函数还有一种特殊形式，通过只考虑新进入窗口的数据和离开窗

口的数据，让 Spark 增量计算归约结果这种特殊形式需要提供归约函数的一个逆函数，比如 + 对应的逆函数为 -。对于较大的窗口，提供逆函数可以大大提高执行效率



//不理解

class ExtractIp extends PairFunction<ApacheAccessLog, String, Long> {
public Tuple2<String, Long> call(ApacheAccessLog entry) {
return new Tuple2(entry.getIpAddress(), 1L);
}
}
class AddLongs extends Function2<Long, Long, Long>() {
public Long call(Long v1, Long v2) { return v1 + v2; }
}
class SubtractLongs extends Function2<Long, Long, Long>() {
public Long call(Long v1, Long v2) { return v1 - v2; }
}
JavaPairDStream<String, Long> ipAddressPairDStream = accessLogsDStream.mapToPair(
new ExtractIp());
JavaPairDStream<String, Long> ipCountDStream = ipAddressPairDStream.
reduceByKeyAndWindow(
new AddLongs(), // 加上新进入窗口的批次中的元素
new SubtractLongs()
// 移除离开窗口的老批次中的元素
Durations.seconds(30), // 窗口时长
Durations.seconds(10)); // 滑动步长

DStream 还提供了 countByWindow() 和 countByValueAndWindow() 作为对数据进行

计数操作的简写。countByWindow() 返回一个表示每个窗口中元素个数的 DStream，而countByValueAndWindow() 返回的 DStream 则包含窗口中每个值的个数

//Java 中的窗口计数操作
JavaDStream<String> ip = accessLogsDStream.map(
new Function<ApacheAccessLog, String>() {
public String call(ApacheAccessLog entry) {
return entry.getIpAddress();
}});
JavaDStream<Long> requestCount = accessLogsDStream.countByWindow(
Dirations.seconds(30), Durations.seconds(10));
JavaPairDStream<String, Long> ipAddressRequestCount = ip.countByValueAndWindow(
Dirations.seconds(30), Durations.seconds(10));

UpdateStateByKey转化操作

DStream 中跨批次维护状态

updateStateByKey() 提供了对一个状态变量的访问，用于键值对形式的DStream。给定一个由（键，事件）对构成的 DStream，并传递一个指定如何根据新的事件更新每个键对应状态的函数， 它可以构建出一个新的 DStream，其内部数据为（键，状态）对

例如，在网络服务器日志中，事件可能是对网站的访问，此时键是用户的 ID。 使用
updateStateByKey() 可以跟踪每个用户最近访问的 10 个页面。这个列表就是“状态”对
象，我们会在每个事件到来时更新这个状态

updateStateByKey()，提供了一个 update(events, oldState) 函数，接收与某键相关

的事件以及该键之前对应的状态，返回这个键对应的新状态

events：是在当前批次中收到的事件的列表（可能为空）

oldState：是一个可选的状态对象，存放在 Option 内；如果一个键没有之前的状态，

这个值可以空缺。

newState：由函数返回，也以 Option 形式存在；我们可以返回一个空的 Option 来表示Spark

想要删除该状态

//用 updateStateByKey() 来跟踪日志消息中各 HTTP 响应代码的计数。这
里的键是响应代码，状态是代表各响应代码计数的整数，事件则是页面访问
//“无限增长”的计数
class UpdateRunningSum implements Function2<List<Long>,
Optional<Long>, Optional<Long>> {
public Optional<Long> call(List<Long> nums, Optional<Long> current) {
long sum = current.or(0L);
return Optional.of(sum + nums.size());
}
};
JavaPairDStream<Integer, Long> responseCodeCountDStream = accessLogsDStream.mapToPair(
new PairFunction<ApacheAccessLog, Integer, Long>() {
public Tuple2<Integer, Long> call(ApacheAccessLog log) {
return new Tuple2(log.getResponseCode(), 1L);
}})
.updateStateByKey(new UpdateRunningSum());

输出操作

输出操作指定了对流数据经转化操作得到的数据所要执行的操作（例如把结果推入外部数

据库或输出到屏幕上）

与 RDD 中 的 惰 性 求 值 类 似， 如 果 一 个 DStream 及 其 派 生 出 的 DStream都 没 有 被 执 行 输 出 操 作， 那 么 这 些 DStream 就 都 不 会 被 求 值。 如 果StreamingContext 中没有设定输出操作，整个 context 就都不会启动

常用的一种调试性输出操作是 print()，它会在每个批次中抓取 DStream 的前十个元素打

印出来

Spark Streaming 对于 DStream 有与 Spark 类似的 save() 操作，它们接受一个目录作为参数来存储文件，还支持通过可选参数来设置文件的后缀名。 每个批次的结果被保存在给定目录的子目录中，且文件名中含有时间和后缀名

//保存为文本文件
ipAddressRequestCount.saveAsTextFiles("outputDir", "txt")
//更为通用的 saveAsHadoopFiles() 函数，接收一个 Hadoop 输出格式作为参数
JavaPairDStream<Text, LongWritable> writableDStream = ipDStream.mapToPair(
new PairFunction<Tuple2<String, Long>, Text, LongWritable>() {
public Tuple2<Text, LongWritable> call(Tuple2<String, Long> e) {
return new Tuple2(new Text(e._1()), new LongWritable(e._2()));
}});
class OutFormat extends SequenceFileOutputFormat<Text, LongWritable> {};
writableDStream.saveAsHadoopFiles(
"outputDir", "txt", Text.class, LongWritable.class, OutFormat.class);
//通用的输出操作 foreachRDD()，它用来对 DStream 中的 RDD 运行任意计
算。这和 transform() 有些类似，都可以让我们访问任意 RDD
ipAddressRequestCount.foreachRDD { rdd =>
rdd.foreachPartition { partition =>
// 打开到存储系统的连接（比如一个数据库的连接）
partition.foreach { item =>
// 使用连接把item存到系统中
}
// 关闭连接
}
}

foreachRDD中可以重用在spark中实现的行动操作

常见的用例之一是把数据写到诸如MySQL 的外部数据库中。对于这种操作， Spark 没有提供对应的 saveAs() 函数，但可以使用 RDD 的 eachPartition() 方法来把它写出去。为了方便， foreachRDD() 也可以提供给我们当前批次的时间，允许我们把不同时间的输出结果存到不同的位置

输入源

一些“核心”数据源已经被打包到 Spark Streaming 的 Maven 工件中，而其他的一些则可以通过 spark-streaming-kafka 等附加工件获取

核心数据源

所有用来从核心数据源创建 DStream 的方法都位于 StreamingContext 中

文件流

因为 Spark 支持从任意 Hadoop 兼容的文件系统中读取数据，所以 Spark Streaming 也就支

持从任意 Hadoop 兼容的文件系统目录中的文件创建数据流

日志始终要复制到 HDFS 上的

要让 Spark Streaming

来处理数据， 我们需要为目录名字提供统一的日期格式，文件也必须原子化创建（比如把

文件移入 Spark 监控的目录）

原子化表示整个操作一次完成。如果 Spark Streaming 将要处理文件时，更多的数据出现了， SparkStreaming 就会无法注意到新添加的数据， 因此原子化在这里很重要。在文件系统中，文件重命名操作一般是原子化的。

JavaDStream<String> logData = jssc.textFileStream(logsDirectory);

除了文本数据， 也可以读入任意 Hadoop 输入格式。只需要将Key、 Value 以及 InputFormat 类提供给 Spark Streaming 即可

//如果先前已经有了一个流处理作业来处理日志， 并已经将得到的每个时间区间内传输的数据分别存储成了一个SequenceFile
ssc.fileStream[LongWritable, IntWritable,
SequenceFileInputFormat[LongWritable, IntWritable]](inputDirectory).map {
case (x, y) => (x.get(), y.get())
}

Akka actor流

附加数据源

用附加数据源接收器来从一些知名数据获取系统中接收的数据，这些接收器都作为 Spark Streaming 的组件进行独立打包了。它们仍然是 Spark 的一部分，不过你需要在构建文件中添加额外的包才能使用它们

Twitter、 Apache、Kafka、 Amazon Kinesis、 Apache Flume

多数据源与集群规模

使用类似 union() 这样的操作将多个 DStream 合并。通过这些操作符，可以把多个输入的 DStream 合并起来

使用多个接收器对于提高聚合操作中的数据获取的吞吐量非常必要（如果只用一个接收器，可能会成为性能瓶颈）

有时需要用不同的接收器来从不同的输入源中接收各种数据，然后使用 join 或 cogroup 进行整合

理解接收器是如何在 Spark 集群中运行的，对于我们使用多个接收器至关重要。每个接收
器都以 Spark 执行器程序中一个长期运行的任务的形式运行，因此会占据分配给应用的
CPU 核心。 此外，我们还需要有可用的 CPU 核心来处理数据。这意味着如果要运行多个
接收器，就必须至少有和接收器数目相同的核心数，还要加上用来完成计算所需要的核心
数。 例如，如果我们想要在流计算应用中运行 10 个接收器，那么至少需要为应用分配 11
个 CPU 核心

不要在本地模式下把主节点配置为 "local" 或 "local[1]" 来运行 Spark
Streaming 程序。 这种配置只会分配一个 CPU 核心给任务，如果接收器运
行在这样的配置里， 就没有剩余的资源来处理收到的数据了。至少要使用
"local[2]" 来利用更多的核心

24/7不间断运行

Spark Streaming 的一大优势在于它提供了强大的容错性保障

只要输入数据存储在可靠的系统中， Spark Streaming 就可以根据输入计算出正确的结果，提供“精确一次”执行的语义（就好像所有的数据都是在没有任何节点失败的情况下处理的一样），即使是工作节点或者驱动器程序发生了失败

要不间断运行 Spark Streaming 应用，需要一些特别的配置。第一步是设置好诸如 HDFS 等可靠存储系统中的检查点机制。不仅如此，我们还需要考虑驱动器程序的容错性（需要特别的配置代码）以及对不可靠输入源的处理

检查点机制

检查点机制是我们在 Spark Streaming 中用来保障容错性的主要机制

使 Spark Streaming 阶段性地把应用数据存储到诸如 HDFS 这样的可靠存储系统中，以供恢复时使用

为以下两个目的服务

控制发生失败时需要重算的状态数。 Spark Streaming 可以通过转化图的谱系图来重算状态，检查点机制则可以控制需要在转化图中回溯多远。

提供驱动器程序容错。 如果流计算应用中的驱动器程序崩溃了，你可以重启驱动器程序

并让驱动器程序从检查点恢复， 这样 Spark Streaming 就可以读取之前运行的程序处理

数据的进度，并从那里继续

所以，检查点机制对于任何生产环境中的流计算应用都至关重要。你可以通过向ssc.checkpoint() 方法传递一个路径参数（HDFS、 S3 或者本地路径均可）来配置检查点机制

ssc.checkpoint("hdfs://...")
即便是在本地模式下，如果你尝试运行一个有状态操作而没有打开检查点机制，Spark Streaming 也会给出提示

驱动器程序容错

驱动器程序的容错要求我们以特殊的方式创建 StreamingContext。我们需要把检查

点 目 录 提 供 给 StreamingContext。 与 直 接 调 用 new StreamingContext 不 同， 应 该 使 用StreamingContext.getOrCreate() 函数

JavaStreamingContextFactory fact = new JavaStreamingContextFactory() {
public JavaStreamingContext call() {
...
JavaSparkContext sc = new JavaSparkContext(conf);
// 以1秒作为批次大小创建StreamingContext
JavaStreamingContext jssc = new JavaStreamingContext(sc, Durations.seconds(1));
jssc.checkpoint(checkpointDir);
return jssc;
}};
JavaStreamingContext jssc = JavaStreamingContext.getOrCreate(checkpointDir, fact);

假设检查点目录不存在，调用工厂函数时把目录创建出来

除了用 getOrCreate() 来实现初始化代码以外，你还需要编写在驱动器程序崩溃时重启驱

动器进程的代码

在大多数集群管理器中， Spark 不会在驱动器程序崩溃时自动重启驱动器进程，所以你需要使用诸如 monit 这样的工具来监视驱动器进程并进行重启。最佳的实现方式往往取决于你的具体环境

//监管模式启动驱动器程序

工作节点容错

为了应对工作节点失败的问题， Spark Streaming 使用与 Spark 的容错机制相同的方法。所

有从外部数据源中收到的数据都在多个工作节点上备份。 所有从备份数据转化操作的过程

中创建出来的 RDD 都能容忍一个工作节点的失败，因为根据 RDD 谱系图，系统可以把丢

失的数据从幸存的输入数据备份中重算出来

接收器容错

运行接收器的工作节点的容错也是很重要的。如果这样的节点发生错误， Spark Streaming

会在集群中别的节点上重启失败的接收器。 然而，这种情况会不会导致数据的丢失取决于

数据源的行为（数据源是否会重发数据）以及接收器的实现（接收器是否会向数据源确认

收到数据）

举个例子， 使用 Flume 作为数据源时，两种接收器的主要区别在于数据丢失

时的保障。在“接收器从数据池中拉取数据”的模型中， Spark 只会在数据已经在集群中

备份时才会从数据池中移除元素。 而在“向接收器推数据”的模型中，如果接收器在数据

备份之前失败， 一些数据可能就会丢失。总的来说，对于任意一个接收器，你必须同时考

虑上游数据源的容错性（是否支持事务）来确保零数据丢失

接收器提供以下保证

所有从可靠文件系统中读取的数据（比如通过 StreamingContext.hadoopFiles 读取的）

都是可靠的， 因为底层的文件系统是有备份的。 Spark Streaming 会记住哪些数据存放到

了检查点中，并在应用崩溃后从检查点处继续执行

对于像 Kafka、推式 Flume、 Twitter 这样的不可靠数据源， Spark 会把输入数据复制到其

他节点上，但是如果接收器任务崩溃， Spark 还是会丢失数据

确保所有数据都被处理的最佳方式是使用可靠的数据源（例如 HDFS、 拉式

Flume 等）。如果你还要在批处理作业中处理这些数据， 使用可靠数据源是最佳方式，因为

这种方式确保了你的批处理作业和流计算作业能读取到相同的数据， 因而可以得到相同的

结果

处理保证

由于 Spark Streaming 工作节点的容错保障， Spark Streaming 可以为所有的转化操作提供

“精确一次”执行的语义， 即使一个工作节点在处理部分数据时发生失败，最终的转化结

果（即转化操作得到的 RDD）仍然与数据只被处理一次得到的结果一样。

然而，当把转化操作得到的结果使用输出操作推入外部系统中时，写结果的任务可能因故

障而执行多次， 一些数据可能也就被写了多次。由于这引入了外部系统，因此我们需要专

门针对各系统的代码来处理这样的情况。 我们可以使用事务操作来写入外部系统（即原子

化地将一个 RDD 分区一次写入），或者设计幂等的更新操作（即多次运行同一个更新操作

仍生成相同的结果）。 比如 Spark Streaming 的 saveAs…File 操作会在一个文件写完时自动

将其原子化地移动到最终位置上，以此确保每个输出文件只存在一份

批次和窗口大小

Spark Streaming 可以使用的最小批次间隔是多少。总的来说， 500 毫秒已

经被证实为对许多应用而言是比较好的最小批次大小。 寻找最小批次大小的最佳实践是从

一个比较大的批次大小（10 秒左右）开始，不断使用更小的批次大小。如果 Streaming 用

户界面中显示的处理时间保持不变， 你就可以进一步减小批次大小。如果处理时间开始增

加，你可能已经达到了应用的极限

相似地，对于窗口操作，计算结果的间隔（也就是滑动步长）对于性能也有巨大的影响。

当计算代价巨大并成为系统瓶颈时，就应该考虑提高滑动步长了

并行度

减少批处理所消耗时间的常见方式还有提高并行度。有以下三种方式可以提高并行度

增加接收器数目

有时如果记录太多导致单台机器来不及读入并分发的话， 接收器会成为系统瓶颈。这时

你就需要通过创建多个输入 DStream（这样会创建多个接收器）来增加接收器数目，然

后使用 union 来把数据合并为一个数据源。

将收到的数据显式地重新分区

如果接收器数目无法再增加， 你可以通过使用 DStream.repartition 来显式重新分区输入流（或者合并多个流得到的数据流）来重新分配收到的数据。

提高聚合计算的并行度

对于像 reduceByKey() 这样的操作，你可以在第二个参数中指定并行度，我们在介绍

RDD 时提到过类似的手段

垃圾回收和内存使用

Java 的垃圾回收机制（简称 GC）也可能会引起问题。你可以通过打开 Java 的并发标志—

清除收集器（Concurrent Mark-Sweep garbage collector）来减少 GC 引起的不可预测的长暂

停。并发标志—清除收集器总体上会消耗更多的资源，但是会减少暂停的发生

通过在配置参数 spark.executor.extraJavaOptions 中添加 -XX:+UseConcMarkSweepGC

来控制选择并发标志—清除收集器

//打开并发标志—清除收集器
spark-submit --conf spark.executor.extraJavaOptions=-XX:+UseConcMarkSweepGC App.jar

除了使用较少引发暂停的垃圾回收器，你还可以通过减轻 GC 的压力来大幅度改善性能。

把 RDD 以序列化的格式缓存（而不使用原生的对象）也可以减轻 GC 的压力，这也是为

什么默认情况下 Spark Streaming 生成的 RDD 都以序列化后的格式存储。使用 Kryo 序列化

工具可以进一步减少缓存在内存中的数据所需要的内存大小

Spark 也允许我们控制缓存下来的 RDD 以怎样的策略从缓存中移除。默认情况下， Spark

使用 LRU 缓存。如果你设置了 spark.cleaner.ttl， Spark 也会显式移除超出给定时间范围

的老 RDD。主动从缓存中移除不大可能再用到的 RDD，可以减轻 GC 的压力