什么是唯一一次
唯一一次，也叫ECO（Exactly once）语义是流式处理程序的重要话题，也是很多流式处理程序问题的关键所在。
为了保证每个消息在系统或网络故障的情况下，被处理一次且仅处理一次。这首先需要流式处理框架需要提供这种功能的支持，然后在消息传递、数据输出存储（比如到MySQL落库）以及具体的代码编写等等各个方面整体考虑和配合，才最终能够保证消息的ECO。
GIT上面有个不错的项目，里面有一些相关的样例代码https://github.com/jizhang/spark-sandbox
一个简单的样例代码
下面的代码实现了从Kafka获取访问数据，然后统计错误日志数量，然后每分钟一次把统计的结果插入到Mysql数据库。
源数据样例：
2017-07-30 14:09:08 ERROR some message
2017-07-30 14:09:20 INFO some message
2017-07-30 14:10:50 ERROR some message
Mysql结果表：
create table error_log (
log_time datetime primary key, -- 具体到分钟粒度，来自源数据中到秒的时间值
log_count int not null default 0 -- 行数
);
部分关键代码：
// 初始化MySQL 连接池
ConnectionPool.singleton("jdbc:mysql://localhost:3306/spark", "root", "")
// 初始化SparkStreaming 的context
val conf = new SparkConf().setAppName("ExactlyOnce").setIfMissing("spark.master", "local[2]")
val ssc = new StreamingContext(conf, Seconds(5))
// 通过Direct API连接Kafka，这种方式方便我们控制偏移量
val messages = KafkaUtils.createDirectStream[String, String](ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](Seq("alog"), kafkaParams))
//遍历 DSStream中的每个RDD，每个RDD都是一个SparkStreaming 的 微批次(micro batch)
messages.foreachRDD { rdd =>
// 数据统计，具体逻辑
// 取源数据中的日期和日志类型，然后过滤ERROR日志，然后截取日期到分钟粒度，然后统计行数入库
val result = rdd.map(_.value)
.flatMap(parseLog) // utility function to parse log line into case class
.filter(_.level == "ERROR")
.map(log => log.time.truncatedTo(ChronoUnit.MINUTES) -> 1)
.reduceByKey(_ + _)
.collect() //这里需要特别注意，数据全部通过collect() 方法回到了Driver进程中，不在各个分区
// 在Driver中对每条数据进行Merge插入，相同时间的数据更新，不存在的时间则插入。
DB.autoCommit { implicit session =>
result.foreach { case (time, count) =>
sql"""
insert into error_log (log_time, log_count)
value (${time}, ${count})
on duplicate key update log_count = log_count + values(log_count)
""".update.apply()
}
}
}
流式处理语义
流式处理，一共分为三个阶段，分别是 接收数据、处理数据和输出数据
这三个阶段都需要相互配合来保证ECO。
接收数据
接收数据过程，比较依赖具体的数据源。不同的数据源有不同的处理方式来保证每条消息的ECO。
比如，对HDFS这种可容错的文件系统，可以和SparkStreaming 整合保证ECO
对于有反馈的消息队列，如 RabbitMQ，可以结合 Spark 的 Write Ahead Log 来保证ECO
对于不可靠的接收器，比如 SocketTextStream，很容易因为Driver的故障导致数据丢失。
对于Kakfa，通过Direct API，也就是直接操作偏移量的，也可以提供ECO支持
处理数据
SparkStreaming 通过 DSStream，底层是一个个连续的Spark 微批次 RDD。对该数据结构操作，本身就是直接提供ECO支持的。
因此在数据处理这块，即RDD操作这块，本身就可以提供错误重启和ECO。
输出数据
默认SparkStreaming提供At Least Once的输出。当某个Excutor正在执行数据输出时发生故障，Driver重新启动一个Excutor，然后会再次进行数据输出。
这个过程，可能回导致同样的数据多次输出。
有两种方式可以解决这种重复输出的问题
- 基于幂等的更新方式
- 基于事务的更新方式
幂等输出和ECO
对于固定的一份源数据，重复经过相同逻辑处理并进行写入操作，且产生相同的结果。一般这种操作就被认为是幂等更新。
举个例子，saveAsTextFile操作，被认为是典型的幂等操作。多次对一个RDD和同一个文件调用该操作，结果值是固定不变的。上次操作和下次操作之间是互不相关的。
如果message是由 unique key的，那么message在写入database的时候，可以通过相同key执行update，不存在的key执行insert，来实现幂等更新。
但是，幂等更新一般适合于map型的sparkStreaming逻辑。举个例子：
一个message进入sparkStreaming计算引擎，经过一系列计算，最终出来的结果也是一个message，且这个message的一个或者两个字段是unique的。这种类型的处理可以通过insert on duplicate update的语法保证幂等更新。
如果sparkStreaming执行了reduceByKey或者Count等操作，这个时候，比较难以通过这种方式保证幂等更新。
幂等更新需要对kafka做一些配置
- 设置enable.auto.commit为false，默认情况下，DStream读取kafka数据时，会在接收到数据后，立马就提交offsets。把这个参数设置为false之后，是为了能够在一个批次执行完成后再提交offset.
- 打开SparkStreaming的checkpoint进行kafka 的 offsets存储，不过，如果更改过了代码之后，checkpoint会失效，因此，还需要通过下面的方式进一步的存储offsets作为补充：
- 结果输出到target以后，再提交offsets。kafka提供了commitAsync API，用来异步提交offsets。同时，提供了HasOffsetRangers类来从RDD中抽取kafka的Offsets值。
messages.foreachRDD { rdd =>
// 抽取RDD中的kafka offsets，以每个微批次（micro btch）RDD为单位进行获取
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
rdd.foreachPartition { iter =>
//遍历微批次RDD的每个分区，然后输出都数据库
//注意，这次没有把所有数据collect()到Driver后再输出
// 注意，因为这次插入操作时幂等插入，类似于 insert or update on duplicate key这种，
//所以即使部分分区失败，或者整个任务失败重启，都不会产生多余错误的数据
}
//把每个分区的数据插入到数据库之后，再提交
//如果之前输出出现任何问题，则不会提交偏移量，那么则会从之前的偏移量重新计算
messages.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}
这里如果有人由疑问：如果某些个分区输出到数据库失败重跑，或者，整个任务挂掉需要重跑，会不会导致重复数据？或者一个数据被计算多次？
答案是：不会
要注意，上一段代码，是基于幂等输出的结果之上的，比如说，saveAsTextFile，或者是对数据库进行 insert on duplicate key update更新。
所以，对于一个微批次来说，不管失败多少次，只要重启后的微批次还是通过之前的offsets去kafka取数，计算的结果永远是一致的，那么输出的结果也永远是一致的，那么通过幂等更新的方式，最终数据库中或者结果文件中的内容也是一致的！！！
幂等更新天生对于重复型输出可以保证ECO
事务输出和ECO
幂等型更新，很多时候是不能保证的，因为我们会对流式数据进行各种维度的统计和计算，那么这个时候，就没法保证数据的幂等更新。
事务型输出，对于map型计算和aggregation类型的计算都有很好的支持，基本原理就是在同一个事务中，同时执行数据输出和更新kafka offsets，做到一起成功或者一起失败。
在SparkStreaming的模型中，offset的粒度是到partition粒度的，也就是一个分区都有自己的一个offsets信息。然后对于一个RDD来说，所有分区的offset范围之和，就是RDD的offsets范围。
对于map型操作，因为SparkStreaming的一个分区对应一个kafka的分区，因此我们可以获取到每个分区的offsets信息。这样，可以在输出时，做到更细粒度的控制。
messages.foreachRDD { rdd =>
//获取整个RDD的offsets信息
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
rdd.foreachPartition { iter =>
// 获取到每个分区粒度的offsets信息，做更细力度的事务控制。之前事务都是控制到RDD级别的offsets记录，现在可以控制到partition级别的记录
val offsetRange = offsetRanges(TaskContext.get.partitionId)
}
}
但是map型操作毕竟还是比较少的，更多的是aggregate类型的操作，这种操作设计到shffule。经过shuffle以后，RDD的partition就和Kafka的patition没有对应关系了。因此这个时候，就需要对整个RDD放在一个事务中进行数据插入和offset更新
messages.foreachRDD { rdd =>
//获取整个RDD的所有offsets信息信息
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
// 对RDD进行aggregate操作，并返回结果RDD
val result = processLogs(rdd).collect() // parse log and count error
// 开启一个事务（一个RDD就对应一个事务）
DB.localTx { implicit session =>
result.foreach { case (time, count) =>
// 对RDD结果值进行入库，注意，此刻还未提交事务，下一步需要提交各分区的offsets
}
// 取出各个partition的offsets信息，并且按照 topic + partition 作为唯一键，进行offsets的更新存储
// kafka在建立topic的时候，就已经指定好了分区个数
// 此刻取到的分区partiton就是对应的是kafka topic的分区
offsetRanges.foreach { offsetRange =>
val affectedRows = sql"""
update kafka_offset set offset = ${offsetRange.untilOffset}
where topic = ${topic} and `partition` = ${offsetRange.partition}
and offset = ${offsetRange.fromOffset}
""".update.apply()
if (affectedRows != 1) {
// 如果更新失败，说明程序出现问题，需要抛出异常
throw new Exception("fail to update offset")
}
}
}
}
结尾
Exactly-once 是流处理中非常强的语义，不可避免地会给你的应用带来一些开销，影响吞吐量。 它也不适用于窗口操作。 因此，您需要决定是否有必要花费这些精力，或者即使数据丢失很少，使用较弱的语义也足够了。 但肯定知道如何实现精确一次是一个很好的学习机会，而且非常有趣。
网友评论