有状态的计算作为容错以及数据一致性的保证,是当今实时计算必不可少的特性之一,流行的实时计算引擎包括 Google Dataflow、Flink、Spark (Structure) Streaming、Kafka Streams 都分别提供对内置 State 的支持。State 的引入使得实时应用可以不依赖外部数据库来存储元数据及中间数据,部分情况下甚至可以直接用 State 存储结果数据,这让业界不禁思考: State 和 Database 是何种关系?有没有可能用 State 来代替数据库呢?
在这个课题上,Flink 社区是比较早就开始探索的。总体来说,Flink 社区的努力可以分为两条线: 一是在作业运行时通过作业查询接口访问 State 的能力,即 QueryableState;二是通过 State 的离线 dump 文件(Savepoint)来离线查询和修改 State 的能力,即即将引入的 Savepoint Processor API。
QueryableState
在 2017 年发布的 Flink 1.2 版本,Flink 引入了 QueryableState 的特性以允许用户通过特定的 client 查询作业 State 的内容 [1],这意味着 Flink 应用可以在完全不依赖 State 存储介质以外的外部存储的情况下提供实时访问计算结果的能力。
只通过 Queryable State 提供实时数据访问
然而,QueryableState 虽然设想上比较理想化,但由于依赖底层架构的改动较多且功能也比较受限,它一直处于 Beta 版本并不能用于生产环境。针对这个问题,在前段时间腾讯的工程师杨华提出 QueryableState 的改进计划 [2]。在邮件列表中,社区就 QueryableState 是否可以用于代替数据库作了讨论并出现了不同的观点。笔者结合个人见解将 State as Database 的主要优缺点整理如下。
优点:
更低的数据延迟。一般情况下 Flink 应用的计算结果需要同步到外部的数据库,比如定时触发输出窗口计算结果,而这种同步通常是定时的会带来一定的延迟,导致计算是实时的而查询却不是实时的尴尬局面,而直接 State 则可以避免这个问题。
相对地,假如用 Operator State 来记录总得分和总时长(并行度设为 1),我们注册 total_score 和 total_time 两个 State,得到的表有两个:
total_score |
------- |
14,500 |
total_time5,600
至此 Savepoint 和 Database 的对应关系应该是比较清晰明了的。而对于 Savepoint 来说还有不同的 StateBackend 来决定 State 具体如何持续化,这显然对应的是数据库的存储引擎。在 MySQL 中,我们可以通过简单的一行命令 ALTER TABLE xxx ENGINE = InnoDB; 来改变存储引擎,在背后 MySQL 会自动完成繁琐的格式转换工作。而对于 Savepoint 来说,由于 StateBackend 各自的存储格式不兼容,目前尚不能方便地切换 StateBackend。为此,社区在不久前创建 FLIP-41 [5] 来进一步完善 Savepoint 的可操作性。
总结
State as Database 是实时计算发展的大趋势,它并不是要代替数据库的使用,而是借鉴数据库领域的经验拓展 State 接口使其操作方式更接近我们熟悉的数据库。对于 Flink 而言,State 的外部使用可以分为在线的实时访问和离线的访问和修改,分别将由 Queryable State 和 Savepoint Processor API 两个特性支持。