Spark Streaming中流式计算的困境与解决之道

Spark streaming 在各种流程处理框架生态中占着举足轻重的位置，但是不可避免地也会面对网络波动带来的数据延迟的问题，所以必须要进行增量数据的累加。在更新Spark 应用的时候或者其他不可避免的异常宕机的时候，增量累加会带来重复消费的问题，在一些需要严格保证 exact once 的场景下，这个时候我们就需要进行离线修复，从而保证exact once 语义，本文将针对这个问题，提供一些常见的解决方案和处理方式。

下图中展示了数据延迟的一个场景：

在讨论解决消息乱序问题之前，需先定义时间和顺序。在流处理中，时间的概念有两个：

Event time ：Event time是事件发生的时间，经常以时间戳表示，并和数据一起发送。带时间戳的数据流有，Web服务日志、监控agent的日志、移动端日志等；
Processing time ：Processing time是处理事件数据的服务器时间，一般是运行流处理应用的服务器时钟。
上图中 time1,time2, time3等是我们Spark straming 拿到消息将要处理的时间，图中方块中的数字代表这个event 产生的时间，有可能因为网络抖动导致部分机器上的日志收集产生了延迟，在time3的batch中包含event time 为2的日志，特别说明一下， kafka 中的不同分区的消息也是没有顺序的。

在实时处理过程中也就产生了两个问题：

Spark streaming 从Kafka 中拉取到的一批数据，我们可能认为里面包含多个时间区间的数据
同一个时间的数据可能出现在多个 batch 中
针对第一个问题，一个 batch 中包含多个时间区间的数据，加入我们的区间粒度是5分钟，那么一个batch钟有可能包含 0~5 时间区间中的部分数据，也有可能包含 5~10 时间区间中的部分数据，这个很好处理，我们先对时间进行向下5分钟取整，然后使用取整后的时间分为多组，然后计算出来指标，` select time, count(*) group by 取整（time），就算出来了这个batch中每个时间区间中的数据。

但是对于第二个问题，就很麻烦，图中举例，时间区间中 2 出现在了 time2 和time3, 我们需要在两个batch中计算出2 的指标，然后进行累计，这个累计的过程，你可以在内存中保存状态，使用Spark streaming 中的 UpdateStateByKey等算子，但是不推荐这样使用，这样就在你的应用中引入了状态和Checkpoint机制，还有一个方法，就是把这个状态放在持久化存储中，比如每次都在 Redis, 或者Hbase 中进行累计，Spark 从 Kafka 拉取日志是可以做到至少消费一次，但是这种模式很难保证 exact once 。

假如有下面一种情形，

就会存在这种情况，我们对 job1 执行 Checkpoint 操作，然后 job1 被调度执行，从Kafka 拉取数据处理，然后结果保存在HBase 中，保存了一半，机器挂了，如果重启，recover, 这时候 job1 就会被重复执行， Kafka 中的数据就会被重复消费， HBase中的部分指标也就多加了一份，虽然我们可以使用 Spark 或者 Flink 中提供的 Watermark 功能。

也就是维护一个窗口，然后设置一个最大等待时间， T1 ~T4 中的数据到了最大等待时间后就会触发计算，但是这样也会有问题，如果部分数据的延迟超过了最大等待时间，这部分数据也就永远的丢失了。

当然如果业务可以容忍，那么使用这个功能也是可以的，每次都使用全量覆盖操作。

解决方案
以上我们面临的问题是 Spark streaming + Kafka 组合可以保证at lease once ，但是很难保证 exact once, 也就是会重复消费，我们得想办法做到去重，计算结果落地存储会有两种模式：

append 增量的模式，也就是每次都做累加
complete的模式，也即是保证幂等性，每次都是覆盖，保证没有副作用
因为同一个时间的数据可能出现在多个 batch 中，所以我们在准实时计算中，只能是append 模式，上文我们已经论证过了，这种模式会出现重复消费的问题。

由于机器挂了的现象是偶发的，所以我们可以在挂掉后，对数据进行离线修复，也就是我们要保证有一份全量的离线数据。

这份数据我们要保证是不漏不多，而且是按照event time 时间区间分开的，这样我们就可以针对出问题的时间区间，加载这个时间区间的离线数据，算出结果，然后进行覆盖。这样就保证了数据的准确性。

我们落地的数据的特点是：

全量的，不漏不多
按照定义的时间区间分片
因为从Kafka 中拉取存储能保证不丢，这里我们考虑如何去重，首先我们要对消息能有一个唯一 ID, 我们使用Kafka的partition加offset作为这个消息的唯一ID，如果存储到HBase, 这样的话在生成一个消息的时候，我们的ID就不会重复，即使你重跑很多次，HBase会自动把它去重。

如果存储到 hdfs, 我们可以每行数据前面都用 ID 作为头字段，离线处理的时候根据这个字段先进行去重处理，这样也能保证了 exact once 语义。
输出流程
我们看下 Spark streaming 存储到HDFS或者HBase 都会调用 saveAsHadoopDataset。

val writer = new SparkHadoopWriter(hadoopConf)
writer.open()
 Utils.tryWithSafeFinallyAndFailureCallbacks {
        while (iter.hasNext) {
          val record = iter.next()
          writer.write(record._1.asInstanceOf[AnyRef], record._2.asInstanceOf[AnyRef])
        }
      }(finallyBlock = writer.close())
writer.commit()

这里根据你传入的 OutFormat 调用 getwriter。

然后再 writer上调用 open write close commit 方法。

这里如果是 HBase 就是调用 HBase client 的写入方法：

用户提交put请求后，HBase客户端会将put请求添加到本地buffer中，符合一定条件就会通过AsyncProcess异步批量提交。HBase默认设置autoflush=true，表示put请求直接会提交给服务器进行处理；用户可以设置autoflush=false，这样的话put请求会首先放到本地buffer，等到本地buffer大小超过一定阈值（默认为2M，可以通过配置文件配置）之后才会提交。很显然，后者采用group commit机制提交请求，可以极大地提升写入性能，但是因为没有保护机制，如果客户端崩溃的话会导致提交的请求丢失。
在提交之前，HBase会在元数据表.meta.中根据rowkey找到它们归属的region server，这个定位的过程是通过HConnection的locateRegion方法获得的。如果是批量请求的话还会把这些rowkey按照HRegionLocation分组，每个分组可以对应一次RPC请求。
HBase会为每个HRegionLocation构造一个远程RPC请求MultiServerCallable，然后通过rpcCallerFactory.newCaller()执行调用，忽略掉失败重新提交和错误处理，客户端的提交操作到此结束。
这里如果是 HDFS 文件写入：

首先根据 TaskAttemptID构造出来一个临时写入路径，构造一个文件流
写入临时写入路径
commit 的时候调用 commitTask 根据目标路径是否存在，如果已经存在就删除临时文件，报错，如果不存在就直接 rename, 把临时文件名，改为目标文件名，这里主要是防止多个分区写入同一个目标文件，导致的冲突。
多文件分组输出
如果有一个需求，需要把数据根据不同的key输出到不同的文件中，上文中，我们先根据 batch 进行分组，然后不同分组的文件输出到不同的文件，这时候就需要用到MultipleOutputFormat

TreeMap<String, RecordWriter<K, V>> recordWriters = new TreeMap<String, RecordWriter<K, V>>();
K actualKey = generateActualKey(key, value);
V actualValue = generateActualValue(key, value);
RecordWriter<K, V> rw = this.recordWriters.get(finalPath);
if (rw == null) {
    rw = getBaseRecordWriter(myFS, myJob, finalPath, myProgressable);
    this.recordWriters.put(finalPath, rw);
}
rw.write(actualKey, actualValue);

这里就是维护了一个TreeMap, 里面每个不同的key, 构造一个 writer, 这个writer 是getBaseRecordWriter -> theTextOutputFormat.getRecordWriter根据临时路径构造出一个输出流，包装为一个 LineRecordWriter 最终的 writer就是在这个 DataOutputStream 上进行输出，

上层多文件输出根据不同的key, 从treeMap上获取到不同的文件输出流，然后进行多文件输出。

这里会存在一个问题，同一个时间的数据可能出现在多个 batch 中，就是会产生很多小文件，HDFS 对小文件支持很差，我们需要合并小文件，但是我们也可以直接在输出的时候进行 append 操作，就直接避免了产生小文件。

这里就需要改源码了。

上面的类图可以清楚的显示类图的关系， MultipleOutputFormat 的writer 会调用子类的 getBaseRecordWriter，我们可以在这里改写一下，使用我们自己的 TextOutputFormatNew 的 getRecordWriterNew 方法，在方法里面构造输出流的时候，如果文件已经存在，就进行 append 操作。

val fileOut: FSDataOutputStream = if (HDFSFileService.existsPath(file)) {
        println("appendfile")
        fs.append(file)
      } else {
        println("createfile")
        fs.create(file, progress)
      }
def getTaskOutputPath(job: JobConf, iname: String): Path = {
    val name: String = job.get(org.apache.hadoop.mapreduce.lib.output.FileOutputFormat.OUTDIR)
    val completePath = name + "/" + iname
    val path = new Path(completePath)
    path
  }

把构造临时路径的方法也修改了，强制不产生临时路径，每次都往同一个文件中进行 append, 这样就达到了目的。

小结
本文提供的解决方案，在不修改Spark 源码本身的前提下，进行了一些必要的扩展，其实本质上来讲，就是我们假定异常状况是经常发生的，我们就要面对它，就是要对输入流 kafka 中的原始数据进行唯一标识，保证可以去重，然后持久化。对发生异常的时间区间，进行数据重放，就像数据中用 redo 日志进行重放一样。

作者介绍
孙彪彪，目前在七牛云存储从事 Spark 相关工作，对 Spark 有深入的研究和实践，微信公众号：Spark技术分享。email： 1319027852@qq.com。

转载：http://www.infoq.com/cn/articles/dilemma-and-solution-streaming-computing?hmsr=toutiao.io&utm_medium=toutiao.io&utm_source=toutiao.io

关注公众号：“程序员成长软技能” ，日拱一卒，功不唐捐！