xiaoyutongxue 2020-05-06
本文摘自书籍《Flink基础教程》
当在分布式系统中引入状态时,自然也引入了一致性问题。一致性实际上是“正确性级别”的另一种说法,即在成功处理故障并恢复之后得到的结果,与没有发生任何故障时得到的结果相比。在流处理中,一致性分为 3 个级别。
曾经, at-least-once 非常流行。第一代流处理器(如 Storm 和 Samza)刚问世时只保证 at-least-once,原因有二:
最先保证 exactly-once 的系统(Storm Trident 和 Spark Streaming)在性能和表现力这两个方面付出了很大的代价:
val mapData: DataStream[(String, Int)] = dataSource.flatMap(f => f.split(" ")).map((_, 1))val mapWithStageData: DataStream[(String, Int)] = mapData.keyBy(_._1).mapWithState((in: (String, Int), count: Option[Int]) => { count match { case Some(c) => ((in._1, c + in._2), Some(c + in._2)) case None => ((in._1, in._2), Some(in._2)) }})
该程序有两个算子:keyBy算子用来将记录按照第一个元素(一个字符串)进行分组,根据该key将数据进行重新分区,然后将记录再发送给下一个算子:有状态的map算子(mapWithState)。map算子在接收到每个元素后,将输入记录的第二个字段的数据加到现有总数中,再将更新过的元素发射出去。
上图表示程序的初始状态:输入流中的6条记录被检查点屏障(checkpoint barrier)隔开,所有的map算子状态均为0(计数还未开始)。所有key为a的记录将被顶层的map算子处理,所有key为b的记录将被中间层的map算子处理,所有key为c的记录则将被底层的map算子处理。其中 ["b",2] 在检查点之前被处理, ["a",2] 则在检查点之后被处理
当该程序处理输入流中的 6 条记录时,涉及的操作遍布 3 个并行实例(节点、 CPU 内核等)。那么,检查点该如何保证 exactly-once 呢?检查点屏障和普通记录类似。它们由算子处理,但并不参与计算,而是会触发与检查点相关的行为。当读取输入流的数据源(在本例中与 keyBy 算子内联)遇到检查点屏障时,它将其在输入流中的位置保存到稳定存储中。如果输入流来自消息传输系统(Kafka 或 MapR Streams),这个位置就是偏移量。 Flink 的存储机制是插件化的,稳定存储可以是分布式文件系统,如HDFS、 S3 或 MapR-FS。下图展示了这个过程。
如下图所示:位于检查点之前的所有记录(["b",2]、 ["b",3] 和 ["c",1])被 map 算子处理之后的情况。此时,稳定存储已经备份了检查点屏障在输入流中的位置(备份操作发生在检查点屏障被输入算子处理的时候)。 map 算子接着开始处理检查点屏障,并触发将状态异步备份到稳定存储中这个动作。检查点屏障像普通记录一样在算子之间流动。当 map 算子处理完前 3 条记录并收到检查点屏障时,它们会将状态以异步的方式写入稳定存储。
当 map 算子的状态备份和检查点屏障的位置备份被确认之后,该检查点操作就可以被标记为完成,如下图所示。我们在无须停止或者阻断计算的条件下,在一个逻辑时间点(对应检查点屏障在输入流中的位置)为计算状态拍了快照。通过确保备份的状态和位置指向同一个逻辑时间点,从而保证 exactly-once。值得注意的是,当没有出现故障时, Flink 检查点的开销极小,检查点操作的速度由稳定存储的可用带宽决定。值得注意的是,备份的状态值与实际的状态值是不同的。备份反映的是检查点的状态
如果检查点操作失败, Flink 会丢弃该检查点并继续正常执行,因为之后的某一个检查点可能会成功。虽然恢复时间可能更长,但是对于状态的保证依旧很有力。只有在一系列连续的检查点操作失败之后, Flink 才会抛出错误,因为这通常预示着发生了严重且持久的错误。
在这种情况下, Flink 会重新拓扑(可能会获取新的执行资源),将输入流倒回到上一个检查点,然后恢复状态值并从该处开始继续计算。
检查点由 Flink 自动生成,用来在故障发生时重新处理记录,从而修正状态。 Flink 用户还可以通过另一个特性有意识地管理状态版本,这个特性叫作保存点(savepoint)
上图中, v.0 是某应用程序的一个正在运行的版本。我们分别在 t1 时刻和 t2时刻触发了保存点。因此,可以在任何时候返回到这两个时间点,并且重启程序。更重要的是,可以从保存点启动被修改过的程序版本。举例来说,可以修改应用程序的代码(假设称新版本为 v.1),然后从 t1 时刻开始运行改动过的代码。这样一来, v.0 和 v.1 这两个版本同时运行,并在之后的时间里获取各自的保存点。
保存点可用于应对流处理作业在生产环境中遇到的许多挑战:
我们已经通过简单的计数例子了解了 Flink 如何保证状态的一致性(即保证exactly-once)。接下来看看端到端的情况,因为在生产环境中可能会部署这种应用程序。
下面的应用程序架构中,有状态的 Flink 应用程序消费来自消息队列的数据,然后将数据写入输出系统,以供查询。底部的详情图展示了 Flink 应用程序的内部情况。
输入数据来自一个分区存储系统(如 Kafka 或者 MapR Streams 这样的消息队列)。底部的详情图展示了 Flink 拓扑,其中包含 3 个算子。source 读取输入数据,根据 key 分区,并将数据路由到有状态的算子实例。有状态的算子将状态内容(比如前例中的计数结果)或者一些衍生结果写入 sink,再由 sink 将结果传送到输出存储系统中(例如文件系统或数据库)。接着,查询服务(比如数据库查询 API)就可以允许用户对状态进行查询(最简单的例子就是查询计数结果),因为状态已经被写入输出存储系统了。
需要记住的是,在本例中,输出反映的是截至最近一次写入状态之时, Flink 应用程序中的状态内容。
在将状态内容传送到输出存储系统的过程中,如何保证 exactly-once 呢?这叫作端到端的一致性。本质上有两种实现方法,用哪一种方法则取决于输出存储系统的类型,以及应用程序的需求:
值得注意的是,这两种方法恰好对应关系数据库系统中的两种为人所熟知的事务隔离级别: 已提交读(read committed)和未提交读(read uncommitted)。已提交读保证所有读取(查询输出)都只读取已提交的数据,而不会读取中间、传输中或“脏”的数据。之后的读取可能会返回不同的结果,因为数据可能已被改变。未提交读则允许读取“脏”数据;换句话说,查询总是看到被处理过的最新版本的数据。
某些应用程序可以接受弱一点的语义,所以 Flink 提供了支持多重语义的多种内置输出算子,如支持未提交读语义的分布式文件输出算子。用户可以根据输出存储系统的能力和应用程序的需求选择合适的语义。根据输出存储系统的类型, Flink 及与之对应的连接器可以一起保证端到端的一致性,并且支持多种隔离级别。
上面的应用程序架构。之所以需要有输出存储系统,是因为外部无法访问 Flink 的内部状态,所以输出存储系统成了查询目标。但是,如果可以直接查询状态,则在某些情况下根本就不需要输出存储系统,因为状态本身就已经包含了查询所需的信息。这种情况在许多应用程序中真实存在, 直接查询状态可以大大地简化架构,同时大幅提升性能