4、周伊莎、马越-Flink SQL上的状态迁移和查询 .pdf

1、周伊莎/马越 字节跳动基础架构工程师State Migration and Query on State Migration and Query on FlinkFlink SQLSQL背景背景SQLSQL 作业的作业的状态迁移状态迁移StateState 的离线的离线查询查询未来展望未来展望#1#1#2#2#3#3#4#4#1#1背景背景状态的作用状态在流式作业的生命周期中扮演了十分重要的角色计算状态在版本间迁移运行时容错计算逻辑正确性校验一些定义：一些定义：SnapshotSnapshot-a global,consistent image of the state of a Flink

2、jobCheckpointCheckpoint-a snapshot taken automatically by FlinkSavepointSavepointa snapshot triggered manually by a userSQL 与状态DataStream 作业SQL 作业可以通过 API 完成对状态的控制：1.注册和使用状态2.查询和修改状态3.设置 UID/UIDHash 以进行状态迁移4.自定义 State Serializer5.RelNodeTransformationSQLJobGraphStreamGraphSQLTableDataStream APIRunti

3、meSQL作业的状态对用户来说是完全黑盒的？SQL 作业使用状态的痛点l状态查询困难状态查询困难n 校验逻辑正确性n 快速定位异常输出来源l状态状态难以难以在迭代版本间迁移在迭代版本间迁移n 简单的字段增减不影响状态恢复n 简单的逻辑变更不影响状态恢复#2#2SQLSQL 作业的状态迁移作业的状态迁移现状状态无法迁移时：1.丢弃旧状态2.回拨 Offset，重跑任务资源浪费产生 LAG长周期任务输入缺失为什么需要迁移旧状态？现状状态迁移的充分必要条件：1.Operator ID 保持不变2.State Serializer 前后兼容问题1:DAG 极易发生变更SourceCalcGroupAg

4、gSink1.隐式修改：修改计算逻辑、打开 minibatch 等2.显式修改：增删维表、source、sink 等为什么 SQL 作业状态难以迁移？现状状态迁移的充分必要条件：1.Operator ID 保持不变2.State Serializer 前后兼容为什么为什么 SQLSQL 作业状态难以迁移？作业状态难以迁移？问题问题2 2:State Serializer 不可兼容1.增删字段2.修改字段的类型AccumulatorAccumulators:ValueStateMAXMAXCOUNTCOUNTSUMSUMSUMSUMMAXMAXCOUNTCOUNTSUMSUMSUMSUMLAST

5、_VALUELAST_VALUEData Talks线上最近30日状态恢复失败 Case 分析由拓扑图发生变更导致的 Case 占比75%由 State Serializer不可兼容导致的Case 占比25%解决思路RelNodeTransformationSQLJobGraphStreamGraphSQLTableDataStream APIRuntimeFlink 在 DataStream API 层为状态迁移提供的能力：设置设置 UID/UID/UIDHashUIDHashSQL 作业如何使用这种能力？1.为 SQL 作业提供的 DAG 预览2.允许用户对 DAG 进行编辑3.将用户编辑

6、的 UID/UIDHash 传递到运行时DAG 预览#1 为什么不直接存储StreamGraph?#2 如何与算子形成稳定映射？复用 StreamGraphHasherV2 来为每个节点生成确定性的 id保持隔离，不被 Flink 迭代影响PlanGraph 抽象FieldExplainid确定性idgeneratedOperatorID与 JobGraph 中算子的OperatorID 对应userProvidedHash用户提供的 UIDHashhasState此算子是否使用状态其他 StreamNode 属性DAG 预览初始可视化效果初始可视化效果DAG 预览任务修改产生 Diff 后整