StreamingContext 是Spark Streaming的入口

flatMap 是一种 “一到多”（one-to-many）的映射算子，它可以将源DStream中每一条记录映射成多条记录 扁平化

ssc.start()            // 启动流式计算

不应该将master参数硬编码到代码里，而是应用通过spark-submit的参数来传递master的值

一旦streamingContext启动，就不能再对其计算逻辑进行添加或修改。

一旦streamingContext被stop掉，就不能restart。

单个JVM虚机同一时间只能包含一个active的StreamingContext。

StreamingContext.stop() 也会把关联的SparkContext对象stop掉，如果不想把SparkContext对象也stop掉，可以将StreamingContext.stop的可选参数 stopSparkContext 设为false。

一个SparkContext对象可以和多个StreamingContext对象关联，只要先对前一个StreamingContext.stop(sparkContext=false)，然后再创建新的StreamingContext对象即可。

任何作用于DStream的算子，其实都会被转化为对其内部RDD的操作。

每个输入DStream（除文件数据流外）都和一个接收器（Receiver – Scala doc, Java doc）相关联，而接收器则是专门从数据源拉取数据到内存中的对象。

如果本地运行Spark Streaming应用，记得不能将master设为”local” 或 “local[1]”。这两个值都只会在本地启动一个线程。而如果此时你使用一个包含接收器（如：套接字、Kafka、Flume等）的输入DStream，那么这一个线程只能用于运行这个接收器，而处理数据的逻辑就没有线程来执行了。因此，本地运行时，一定要将master设为”local[n]”，其中 n > 接收器的个数（有关master的详情请参考Spark Properties）。

将Spark Streaming应用置于集群中运行时，同样，分配给该应用的CPU core数必须大于接收器的总数。否则，该应用就只会接收数据，而不会处理数据。

文件数据流（File Streams）: 可以从任何兼容HDFS API（包括：HDFS、S3、NFS等）的文件系统，创建方式如下：

streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)

Spark Streaming将监视该dataDirectory目录，并处理该目录下任何新建的文件（目前还不支持嵌套目录）。注意：

各个文件数据格式必须一致。

dataDirectory中的文件必须通过moving或者renaming来创建。

一旦文件move进dataDirectory之后，就不能再改动。所以如果这个文件后续还有写入，这些新写入的数据不会被读取。

对于简单的文本文件，更简单的方式是调用 streamingContext.textFileStream(dataDirectory)。

另外，文件数据流不是基于接收器的，所以不需要为其单独分配一个CPU core。

基于自定义Actor的数据流（Streams based on Custom Actors）: DStream可以由Akka actor创建得到，只需调用 streamingContext.actorStream(actorProps, actor-name)。详见自定义接收器（Custom Receiver Guide）。actorStream暂时不支持Python API。

RDD队列数据流（Queue of RDDs as a Stream）: 如果需要测试Spark Streaming应用，你可以创建一个基于一批RDD的DStream对象，只需调用 streamingContext.queueStream(queueOfRDDs)。RDD会被一个个依次推入队列，而DStream则会依次以数据流形式处理这些RDD的数据。

高级数据源在spark-shell中不可用，因此不能用spark-shell来测试基于高级数据源的应用。如果真有需要的话，你需要自行下载相应数据源的Maven工件及其依赖项，并将这些Jar包部署到spark-shell的classpath中。

reduceByKey(func, [numTasks]) 如果源DStream 包含的元素为 (K, V) 键值对，则该算子返回一个新的也包含(K, V)键值对的DStream，其中V是由func聚合得到的。注意：默认情况下，该算子使用Spark的默认并发任务数（本地模式为2，集群模式下由spark.default.parallelism 决定）。你可以通过可选参数numTasks来指定并发任务个数。

transform算子（及其变体transformWith）可以支持任意的RDD到RDD的映射操作。也就是说，你可以用tranform算子来包装任何DStream API所不支持的RDD算子。例如，将DStream每个批次中的RDD和另一个Dataset进行关联（join）操作，这个功能DStream API并没有直接支持。不过你可以用transform来实现这个功能，可见transform其实为DStream提供了非常强大的功能支持。

注意，这里transform包含的算子，其调用时间间隔和批次间隔是相同的。所以你可以基于时间改变对RDD的操作，如：在不同批次，调用不同的RDD算子，设置不同的RDD分区或者广播变量等。

Dstream输出算子可以将DStream的数据推送到外部系统，如：数据库或者文件系统。因为输出算子会将最终完成转换的数据输出到外部系统，因此只有输出算子调用时，才会真正触发DStream transformation算子的真正执行（这一点类似于RDD 的action算子）。

foreachRDD(func) 这是最通用的输出算子了，该算子接收一个函数func，func将作用于DStream的每个RDD上。

func应该实现将每个RDD的数据推到外部系统中，比如：保存到文件或者写到数据库中。

注意，func函数是在streaming应用的驱动器进程中执行的，所以如果其中包含RDD的action算子，就会触发对DStream中RDDs的实际计算过程。

print() 在驱动器（driver）节点上打印DStream每个批次中的头十个元素。

以下列举了一些常见的错误。

通常，对外部系统写入数据需要一些连接对象（如：远程server的TCP连接），以便发送数据给远程系统。因此，开发人员可能会不经意地在Spark驱动器（driver）进程中创建一个连接对象，然后又试图在Spark worker节点上使用这个连接。如下例所示：

dstream.foreachRDD { rdd =>

val connection = createNewConnection()  // 这行在驱动器（driver）进程执行

rdd.foreach { record =>

connection.send(record) // 而这行将在worker节点上执行

}

}

这段代码是错误的，因为它需要把连接对象序列化，再从驱动器节点发送到worker节点。而这些连接对象通常都是不能跨节点（机器）传递的。比如，连接对象通常都不能序列化，或者在另一个进程中反序列化后再次初始化（连接对象通常都需要初始化，因此从驱动节点发到worker节点后可能需要重新初始化）等。解决此类错误的办法就是在worker节点上创建连接对象。

然而，有些开发人员可能会走到另一个极端 – 为每条记录都创建一个连接对象，例如：

dstream.foreachRDD { rdd =>

rdd.foreach { record =>

val connection = createNewConnection()

connection.send(record)

connection.close()

}

}

一般来说，连接对象是有时间和资源开销限制的。因此，对每条记录都进行一次连接对象的创建和销毁会增加很多不必要的开销，同时也大大减小了系统的吞吐量。一个比较好的解决方案是使用 rdd.foreachPartition – 为RDD的每个分区创建一个单独的连接对象，示例如下：

dstream.foreachRDD { rdd =>

rdd.foreachPartition { partitionOfRecords =>

val connection = createNewConnection()

partitionOfRecords.foreach(record => connection.send(record))

connection.close()

}

}

这样一来，连接对象的创建开销就摊到很多条记录上了。

最后，还有一个更优化的办法，就是在多个RDD批次之间复用连接对象。开发者可以维护一个静态连接池来保存连接对象，以便在不同批次的多个RDD之间共享同一组连接对象，示例如下：

dstream.foreachRDD { rdd =>

rdd.foreachPartition { partitionOfRecords =>

// ConnectionPool 是一个静态的、懒惰初始化的连接池

val connection = ConnectionPool.getConnection()

partitionOfRecords.foreach(record => connection.send(record))

ConnectionPool.returnConnection(connection)  // 将连接返还给连接池，以便后续复用之

}

}

注意，连接池中的连接应该是懒惰创建的，并且有确定的超时时间，超时后自动销毁。这个实现应该是目前发送数据最高效的实现方式。

其他要点:

DStream的转化执行也是懒惰的，需要输出算子来触发，这一点和RDD的懒惰执行由action算子触发很类似。特别地，DStream输出算子中包含的RDD action算子会强制触发对所接收数据的处理。因此，如果你的Streaming应用中没有输出算子，或者你用了dstream.foreachRDD(func)却没有在func中调用RDD action算子，那么这个应用只会接收数据，而不会处理数据，接收到的数据最后只是被简单地丢弃掉了。

默认地，输出算子只能一次执行一个，且按照它们在应用程序代码中定义的顺序执行。

你也可以在其他线程里执行SQL查询（异步查询，即：执行SQL查询的线程和运行StreamingContext的线程不同）。不过这种情况下，你需要确保查询的时候 StreamingContext 没有把所需的数据丢弃掉，否则StreamingContext有可能已将老的RDD数据丢弃掉了，那么异步查询的SQL语句也可能无法得到查询结果。举个栗子，如果你需要查询上一个批次的数据，但是你的SQL查询可能要执行5分钟，那么你就需要StreamingContext至少保留最近5分钟的数据：streamingContext.remember(Minutes(5)) （这是Scala为例，其他语言差不多）

对于从网络接收数据的输入数据流（如：Kafka、Flume、socket等），默认的持久化级别会将数据持久化到两个不同的节点上互为备份副本，以便支持容错。

注意，与RDD不同的是，DStream的默认持久化级别是将数据序列化到内存中。

检查点需要保存以下两种数据：

元数据检查点（Metadata checkpointing） – 保存流式计算逻辑的定义信息到外部可容错存储系统（如：HDFS）。主要用途是用于在故障后回复应用程序本身（后续详谈）。元数包括：

Configuration – 创建Streaming应用程序的配置信息。

DStream operations – 定义流式处理逻辑的DStream操作信息。

Incomplete batches – 已经排队但未处理完的批次信息。

数据检查点（Data checkpointing） – 将生成的RDD保存到可靠的存储中。这对一些需要跨批次组合数据或者有状态的算子来说很有必要。在这种转换算子中，往往新生成的RDD是依赖于前几个批次的RDD，因此随着时间的推移，有可能产生很长的依赖链条。为了避免在恢复数据的时候需要恢复整个依赖链条上所有的数据，检查点需要周期性地保存一些中间RDD状态信息，以斩断无限制增长的依赖链条和恢复时间。

总之，元数据检查点主要是为了恢复驱动器节点上的故障，而数据或RDD检查点是为了支持对有状态转换操作的恢复。

如果有以下情况出现，你就必须启用检查点了：

使用了有状态的转换算子（Usage of stateful transformations） – 不管是用了 updateStateByKey 还是用了 reduceByKeyAndWindow（有”反归约”函数的那个版本），你都必须配置检查点目录来周期性地保存RDD检查点。

支持驱动器故障中恢复（Recovering from failures of the driver running the application） – 这时候需要元数据检查点以便恢复流式处理的进度信息。

RDD检查点会增加额外的保存数据的开销。这可能会导致数据流的处理时间变长。因此，你必须仔细的调整检查点间隔时间。如果批次间隔太小（比如：1秒），那么对每个批次保存检查点数据将大大减小吞吐量。另一方面，检查点保存过于频繁又会导致血统信息和任务个数的增加，这同样会影响系统性能。对于需要RDD检查点的有状态转换算子，默认的间隔是批次间隔的整数倍，且最小10秒。开发人员可以这样来自定义这个间隔：dstream.checkpoint(checkpointInterval)。一般推荐设为批次间隔时间的5~10倍。

要运行一个Spark Streaming 应用，你首先需要具备以下条件：

集群以及集群管理器 – 这是一般Spark应用的基本要求，详见 deployment guide。

给Spark应用打个JAR包 – 你需要将你的应用打成一个JAR包。如果使用spark-submit 提交应用，那么你不需要提供Spark和Spark Streaming的相关JAR包。但是，如果你使用了高级数据源（advanced sources – 如：Kafka、Flume、Twitter等），那么你需要将这些高级数据源相关的JAR包及其依赖一起打包并部署。例如，如果你使用了TwitterUtils，那么就必须将spark-streaming-twitter_2.10及其相关依赖都打到应用的JAR包中。

为执行器（executor）预留足够的内存 – 执行器必须配置预留好足够的内存，因为接受到的数据都得存在内存里。注意，如果某些窗口长度达到10分钟，那也就是说你的系统必须知道保留10分钟的数据在内存里。可见，到底预留多少内存是取决于你的应用处理逻辑的。

配置检查点 – 如果你的流式应用需要检查点，那么你需要配置一个Hadoop API兼容的可容错存储目录作为检查点目录，流式应用的信息会写入这个目录，故障恢复时会用到这个目录下的数据。详见前面的检查点小节。

配置驱动程序自动重启 – 流式应用自动恢复的前提就是，部署基础设施能够监控驱动器进程，并且能够在其故障时，自动重启之。不同的集群管理器有不同的工具来实现这一功能：

Spark独立部署 – Spark独立部署集群可以支持将Spark应用的驱动器提交到集群的某个worker节点上运行。同时，Spark的集群管理器可以对该驱动器进程进行监控，一旦驱动器退出且返回非0值，或者因worker节点原始失败，Spark集群管理器将自动重启这个驱动器。详见Spark独立部署指南（Spark Standalone guide）。

YARN – YARN支持和独立部署类似的重启机制。详细请参考YARN的文档。

Mesos – Mesos上需要用Marathon来实现这一功能。

配置WAL（write ahead log）- 从Spark 1.2起，我们引入了write ahead log来提高容错性。如果启用这个功能，则所有接收到的数据都会以write ahead log形式写入配置好的检查点目录中。这样就能确保数据零丢失（容错语义有详细的讨论）。用户只需将 spark.streaming.receiver.writeAheadLog 设为true。不过，这同样可能会导致接收器的吞吐量下降。不过你可以启动多个接收器并行接收数据，从而提升整体的吞吐量（more receivers in parallel）。另外，建议在启用WAL后禁用掉接收数据多副本功能，因为WAL其实已经是存储在一个多副本存储系统中了。你只需要把存储级别设为 StorageLevel.MEMORY_AND_DISK_SER。如果是使用S3（或者其他不支持flushing的文件系统）存储WAL，一定要记得启用这两个标识：spark.streaming.driver.writeAheadLog.closeFileAfterWrite 和 spark.streaming.receiver.writeAheadLog.closeFileAfterWrite。更详细请参考： Spark Streaming Configuration。

设置好最大接收速率 – 如果集群可用资源不足以跟上接收数据的速度，那么可以在接收器设置一下最大接收速率，即：每秒接收记录的条数。相关的主要配置有：spark.streaming.receiver.maxRate，如果使用Kafka Direct API 还需要设置 spark.streaming.kafka.maxRatePerPartition。从Spark 1.5起，我们引入了backpressure的概念来动态地根据集群处理速度，评估并调整该接收速率。用户只需将 spark.streaming.backpressure.enabled设为true即可启用该功能。

性能调优

减少批次处理时间：数据接收并发度 配置多个输入DStream就能从数据源的不同分区分别接收多个数据流。例如，可以将从Kafka拉取两个topic的数据流分成两个Kafka输入数据流，每个数据流拉取其中一个topic的数据，这样一来会同时有两个接收器并行地接收数据，因而增加了总体的吞吐量。同时，另一方面我们又可以把这些DStream数据流合并成一个，然后可以在合并后的DStream上使用任何可用的transformation算子。

接收器的阻塞间隔：该参数由配置参数（configuration parameter）spark.streaming.blockInterval决定。大多数接收器都会将数据合并成一个个数据块，然后再保存到spark内存中。对于map类算子来说，每个批次中数据块的个数将会决定处理这批数据并行任务的个数，每个接收器每批次数据处理任务数约等于 （批次间隔 / 数据块间隔）。例如，对于2秒的批次间隔，如果数据块间隔为200ms，则创建的并发任务数为10。如果任务数太少（少于单机cpu core个数），则资源利用不够充分。如需增加这个任务数，对于给定的批次间隔来说，只需要减少数据块间隔即可。不过，我们还是建议数据块间隔至少要50ms，否则任务的启动开销占比就太高了。

分区数据流：repartition

数据处理并发度

在计算各个阶段（stage）中，任何一个阶段的并发任务数不足都有可能造成集群资源利用率低。例如，对于reduce类的算子，如：reduceByKey 和 reduceByKeyAndWindow，其默认的并发任务数是由 spark.default.parallelism 决定的。你既可以修改这个默认值（spark.default.parallelism），也可以通过参数指定这个并发数量（见PairDStreamFunctions）。

数据序列化

调整数据的序列化格式可以大大减少数据序列化的开销。在spark Streaming中主要有两种类型的数据需要序列化：

输入数据: 默认地，接收器收到的数据是以 StorageLevel.MEMORY_AND_DISK_SER_2 的存储级别存储到执行器（executor）内存中的。也就是说，收到的数据会被序列化以减少GC开销，同时保存两个副本以容错。同时，数据会优先保存在内存里，当内存不足时才吐出到磁盘上。很明显，这个过程中会有数据序列化的开销 – 接收器首先将收到的数据反序列化，然后再以spark所配置指定的格式来序列化数据。

Streaming算子所生产的持久化的RDDs: Streaming计算所生成的RDD可能会持久化到内存中。例如，基于窗口的算子会将数据持久化到内存，因为窗口数据可能会多次处理。所不同的是，spark core默认用 StorageLevel.MEMORY_ONLY 级别持久化RDD数据，而spark streaming默认使用StorageLevel.MEMORY_ONLY_SER 级别持久化接收到的数据，以便尽量减少GC开销。

不管是上面哪一种数据，都可以使用Kryo序列化来减少CPU和内存开销，详见Spark Tuning Guide。另，对于Kryo，你可以考虑这些优化：注册自定义类型，禁用对象引用跟踪（详见Configuration Guide）。

在一些特定的场景下，如果数据量不是很大，那么你可以考虑不用序列化格式，不过你需要注意的是取消序列化是否会导致大量的GC开销。例如，如果你的批次间隔比较短（几秒）并且没有使用基于窗口的算子，这种情况下你可以考虑禁用序列化格式。这样可以减少序列化的CPU开销以优化性能，同时GC的增长也不多。

Spark Streaming应用在集群中占用的内存量严重依赖于具体所使用的tranformation算子。例如，如果想要用一个窗口算子操纵最近10分钟的数据，那么你的集群至少需要在内存里保留10分钟的数据；另一个例子是updateStateByKey，如果key很多的话，相对应的保存的key的state也会很多，而这些都需要占用内存。

DStream持久化级别（Persistence Level of DStreams）: 前面数据序列化（Data Serialization）这小节已经提到过，默认streaming的输入RDD会被持久化成序列化的字节流。相对于非序列化数据，这样可以减少内存占用和GC开销。如果启用Kryo序列化，还能进一步减少序列化数据大小和内存占用量。如果你还需要进一步减少内存占用的话，可以开启数据压缩（通过spark.rdd.compress这个配置设定），只不过数据压缩会增加CPU消耗。

清除老数据（Clearing old data）: 默认情况下，所有的输入数据以及DStream的transformation算子产生的持久化RDD都是自动清理的。Spark Streaming会根据所使用的transformation算子来清理老数据。例如，你用了一个窗口操作处理最近10分钟的数据，那么Spark Streaming会保留至少10分钟的数据，并且会主动把更早的数据都删掉。当然，你可以设置 streamingContext.remember 以保留更长时间段的数据（比如：你可能会需要交互式地查询更老的数据）。

CMS垃圾回收器（CMS Garbage Collector）: 为了尽量减少GC暂停的时间，我们墙裂建议使用CMS垃圾回收器（concurrent mark-and-sweep GC）。虽然CMS GC会稍微降低系统的总体吞吐量，但我们仍建议使用它，因为CMS GC能使批次处理的时间保持在一个比较恒定的水平上。最后，你需要确保在驱动器（通过spark-submit中的–driver-java-options设置）和执行器（使用spark.executor.extraJavaOptions配置参数）上都设置了CMS GC。

其他提示: 如果还想进一步减少GC开销，以下是更进一步的可以尝试的手段：

配合Tachyon使用堆外内存来持久化RDD。详见Spark编程指南（Spark Programming Guide）

使用更多但是更小的执行器进程。这样GC压力就会分散到更多的JVM堆中。

要理解Spark Streaming所提供的容错语义，我们首先需要回忆一下Spark RDD所提供的基本容错语义。

RDD是不可变的，可重算的，分布式数据集。每个RDD都记录了其创建算子的血统信息，其中每个算子都以可容错的数据集作为输入数据。

如果RDD的某个分区因为节点失效而丢失，则该分区可以根据RDD的血统信息以及相应的原始输入数据集重新计算出来。

假定所有RDD transformation算子计算过程都是确定性的，那么通过这些算子得到的最终RDD总是包含相同的数据，而与Spark集群的是否故障无关。

Spark主要操作一些可容错文件系统的数据，如：HDFS或S3。因此，所有从这些可容错数据源产生的RDD也是可容错的。然而，对于Spark Streaming并非如此，因为多数情况下Streaming需要从网络远端接收数据，这回导致Streaming的数据源并不可靠（尤其是对于使用了fileStream的应用）。要实现RDD相同的容错属性，数据接收就必须用多个不同worker节点上的Spark执行器来实现（默认副本因子是2）。因此一旦出现故障，系统需要恢复两种数据：

接收并保存了副本的数据 – 数据不会因为单个worker节点故障而丢失，因为有副本！

接收但尚未保存副本数据 – 因为数据并没有副本，所以一旦故障，只能从数据源重新获取。

此外，还有两种可能的故障类型需要考虑：

Worker节点故障 – 任何运行执行器的worker节点一旦故障，节点上内存中的数据都会丢失。如果这些节点上有接收器在运行，那么其包含的缓存数据也会丢失。

Driver节点故障 – 如果Spark Streaming的驱动节点故障，那么很显然SparkContext对象就没了，所有执行器及其内存数据也会丢失。

Spark Streaming的容错语义。

定义

流式系统的可靠度语义可以据此来分类：单条记录在系统中被处理的次数保证。一个流式系统可能提供保证必定是以下三种之一（不管系统是否出现故障）：

至多一次（At most once）: 每条记录要么被处理一次，要么就没有处理。

至少一次（At least once）: 每条记录至少被处理过一次（一次或多次）。这种保证能确保没有数据丢失，比“至多一次”要强。但有可能出现数据重复。

精确一次（Exactly once）: 每条记录都精确地只被处理一次 – 也就是说，既没有数据丢失，也不会出现数据重复。这是三种保证中最强的一种。

基础语义

任何流式处理系统一般都会包含以下三个数据处理步骤：

数据接收（Receiving the data）: 从数据源拉取数据。

数据转换（Transforming the data）: 将接收到的数据进行转换（使用DStream和RDD transformation算子）。

数据推送（Pushing out the data）: 将转换后最终数据推送到外部文件系统，数据库或其他展示系统。

如果Streaming应用需要做到端到端的“精确一次”的保证，那么就必须在以上三个步骤中各自都保证精确一次：即，每条记录必须，只接收一次、处理一次、推送一次。下面让我们在Spark Streaming的上下文环境中来理解一下这三个步骤的语义：

数据接收: 不同数据源提供的保证不同，下一节再详细讨论。

数据转换: 所有的数据都会被“精确一次”处理，这要归功于RDD提供的保障。即使出现故障，只要数据源还能访问，最终所转换得到的RDD总是包含相同的内容。

数据推送: 输出操作默认保证“至少一次”的语义，是否能“精确一次”还要看所使用的输出算子（是否幂等）以及下游系统（是否支持事务）。不过用户也可以开发自己的事务机制来实现“精确一次”语义。这个后续会有详细讨论。

接收数据语义

不同的输入源提供不同的数据可靠性级别，从“至少一次”到“精确一次”。

从文件接收数据

如果所有的输入数据都来源于可容错的文件系统，如HDFS，那么Spark Streaming就能在任何故障中恢复并处理所有的数据。这种情况下就能保证精确一次语义，也就是说不管出现什么故障，所有的数据总是精确地只处理一次，不多也不少。

基于接收器接收数据

对于基于接收器的输入源，容错语义将同时依赖于故障场景和接收器类型。前面也已经提到过，spark Streaming主要有两种类型的接收器：

可靠接收器 – 这类接收器会在数据接收并保存好副本后，向可靠数据源发送确认信息。这类接收器故障时，是不会给缓存的（已接收但尚未保存副本）数据发送确认信息。因此，一旦接收器重启，没有收到确认的数据，会重新从数据源再获取一遍，所以即使有故障也不会丢数据。

不可靠接收器 – 这类接收器不会发送确认信息，因此一旦worker和driver出现故障，就有可能会丢失数据。

对于不同的接收器，我们可以获得如下不同的语义。如果一个worker节点故障了，对于可靠接收器来书，不会有数据丢失。而对于不可靠接收器，缓存的（接收但尚未保存副本）数据可能会丢失。如果driver节点故障了，除了接收到的数据之外，其他的已经接收且已经保存了内存副本的数据都会丢失，这将会影响有状态算子的计算结果。

为了避免丢失已经收到且保存副本的数，从 spark 1.2 开始引入了WAL（write ahead logs），以便将这些数据写入到可容错的存储中。只要你使用可靠接收器，同时启用WAL（write ahead logs enabled），那么久再也不用为数据丢失而担心了。并且这时候，还能提供“至少一次”的语义保证。

输出算子的语义

输出算子（如 foreachRDD）提供“至少一次”语义保证，也就是说，如果worker故障，单条输出数据可能会被多次写入外部实体中。不过这对于文件系统来说是可以接受的（使用saveAs***Files 多次保存文件会覆盖之前的），所以我们需要一些额外的工作来实现“精确一次”语义。主要有两种实现方式：

幂等更新（Idempotent updates）: 就是说多次操作，产生的结果相同。例如，多次调用saveAs***Files保存的文件总是包含相同的数据。

事务更新（Transactional updates）: 所有的更新都是事务性的，这样一来就能保证更新的原子性。以下是一种实现方式：

用批次时间（在foreachRDD中可用）和分区索引创建一个唯一标识，该标识代表流式应用中唯一的一个数据块。

基于这个标识建立更新事务，并使用数据块数据更新外部系统。也就是说，如果该标识未被提交，则原子地将标识代表的数据更新到外部系统。否则，就认为该标识已经被提交，直接忽略之。