踩坑记| flink state 序列化 java enum 竟然岔劈了

?

感谢您的**「关注  +  点赞 + 再看」**，对博主的肯定，会督促博主持续的输出更多的优质实战内容！！！

?

1.序篇-先说结论

大数据羊说

用数据提升美好事物发生的概率~

30篇原创内容

公众号

本文主要记录博主在生产环境中踩的 flink 针对 java enum serde 时的坑。

结论：在 flink 程序中，如果状态中有存储 java enum，那么添加或者删除 enum 中的一个枚举值时，就有可能导致状态恢复异常，这里的异常可能不是在恢复过程中会实际抛出一个异常，而是有可能是 enum A 的值恢复给 enum B。

我从以下几个章节说明、解决这个问题，希望能抛砖引玉，带给大家一些启发。

1. 踩坑场景篇-这个坑是啥样的

2. 问题排查篇-坑的排查过程

3. 问题原理解析篇-导致问题的机制是什么

4. 避坑篇-如何避免这种问题

5. 总结篇

2.踩坑场景篇-这个坑是啥样的

对任务做一个简单的过滤条件修改，任务重新上线之后，从 flink web ui 确认是从 savepoint 重启成功了，但是实际最终产出的数据上来看却像是没有从 savepoint 重启。

逻辑就是计算分维度的当天累计 pv。代码很简单，在后面会贴出来。

如下图：

2

在 00:04 分重启时出现了当天累计 pv 出现了从零累计的情况。

但是预期正常的曲线应该张下面这样。

1

任务是使用 DataStream 编写（基于 flink 1.13.1）。

public class SenerioTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());env.setParallelism(1);env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);env.addSource(new SourceFunction<SourceModel>() {private volatile boolean isCancel = false;@Overridepublic void run(SourceContext<SourceModel> ctx) throws Exception {// 数据源}@Overridepublic void cancel() {this.isCancel = true;}}).keyBy(new KeySelector<SourceModel, Long>() {@Overridepublic Long getKey(SourceModel value) throws Exception {return value.getUserId() % 1000;}}).timeWindow(Time.minutes(1)).aggregate(new AggregateFunction<SourceModel, Map<Tuple2<DimNameEnum, String>, Long>, Map<Tuple2<DimNameEnum, String>, Long>>() {@Overridepublic Map<Tuple2<DimNameEnum, String>, Long> createAccumulator() {return new HashMap<>();}@Overridepublic Map<Tuple2<DimNameEnum, String>, Long> add(SourceModel value,Map<Tuple2<DimNameEnum, String>, Long> accumulator) {Lists.newArrayList(Tuple2.of(DimNameEnum.province, value.getProvince()), Tuple2.of(DimNameEnum.age, value.getAge()), Tuple2.of(DimNameEnum.sex, value.getSex())).forEach(t -> {Long l = accumulator.get(t);if (null == l) {accumulator.put(t, 1L);} else {accumulator.put(t, l + 1);}});return accumulator;}@Overridepublic Map<Tuple2<DimNameEnum, String>, Long> getResult(Map<Tuple2<DimNameEnum, String>, Long> accumulator) {return accumulator;}@Overridepublic Map<Tuple2<DimNameEnum, String>, Long> merge(Map<Tuple2<DimNameEnum, String>, Long> a,Map<Tuple2<DimNameEnum, String>, Long> b) {return null;}},new ProcessWindowFunction<Map<Tuple2<DimNameEnum, String>, Long>, SinkModel, Long, TimeWindow>() {private transient ValueState<Map<Tuple2<DimNameEnum, String>, Long>> todayPv;@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);this.todayPv = getRuntimeContext().getState(new ValueStateDescriptor<Map<Tuple2<DimNameEnum, String>, Long>>("todayPv", TypeInformation.of(new TypeHint<Map<Tuple2<DimNameEnum, String>, Long>>() {})));}@Overridepublic void process(Long aLong, Context context,Iterable<Map<Tuple2<DimNameEnum, String>, Long>> elements, Collector<SinkModel> out)throws Exception {// 将 elements 数据 merge 到 todayPv 中// 每天零点将 state 清空重新累计// 然后 out#collect 出去即可}});env.execute();}@Data@Builderprivate static class SourceModel {private long userId;private String province;private String age;private String sex;private long timestamp;}@Data@Builderprivate static class SinkModel {private String dimName;private String dimValue;private long timestamp;}enum DimNameEnum {province,age,sex,;}}

3.问题排查篇-坑的排查过程

3.1.愚蠢的怀疑引擎

首先怀疑是状态没有正常恢复。

但是查看 flink web ui 以及 tm 日志，都显示是从 savepoint 正常恢复了。

还怀疑是不是出现了 flink web ui 展示的内容和实际的执行不一致的情况。

但是发现任务的 ck 大小是正常的，复合预期的。

3.2.老老实实打 log 吧

既然能从 savepoint 正常恢复，那么就把状态值用 log 打出来看看到底发生了什么事情呗。

如下列代码，在 ProcessWindowFunction 中加上 log 日志。

this.todayPv.value().forEach(new BiConsumer<Tuple2<DimNameEnum, String>, Long>() {@Overridepublic void accept(Tuple2<DimNameEnum, String> k,Long v) {log.info("key 值：{}，value 值：{}", k.toString(), v);}});

发现结果如下：

...
key 值：(uv_type,男)，value 值：1000
...

发现状态中存储的 DimNameEnum.province，DimNameEnum.age 的数据都是正确的，但是缺缺少了 DimNameEnum.sex，多了 (uv_type,男) 这样的数据，于是查看代码，发现之前多加了一种枚举类型 DimNameEnum.uv_type。代码如下：

enum DimNameEnum {province,age,uv_type,sex,;
}

于是怀疑 flink 针对枚举值的 serde 不是按照枚举值名称来进行匹配的，而是按照枚举值下标来进行匹配的。因此就出现了 DimNameEnum.uv_type 将 DimNameEnum.sex 的位置占了的情况。

4.问题原理解析篇-导致问题的机制是什么

来看看源码吧。

测试代码如下：

public class EnumsStateTest {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env =StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());env.setParallelism(1);env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);TypeInformation<StateTestEnums> t = TypeInformation.of(StateTestEnums.class);EnumSerializer<StateTestEnums> e = (EnumSerializer<StateTestEnums>) t.createSerializer(env.getConfig());DataOutputSerializer d = new DataOutputSerializer(10000);e.serialize(StateTestEnums.A, d);env.execute();}enum StateTestEnums {A,B,C;}
}

debug 结果如下：

首先看看对应的 TypeInformation 和 TypeSerializer。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-FV2uZn7t-1629819487648)(https://mmbiz.qpic.cn/mmbiz_png/DODKOLcDkD2L8ZqGCXriaINgDWnqkLzt36ujZUJDiaGoDTXCaGMpK4u227X8TpvpKxb6aGePQe1BbVjrBUibmP5Ew/640?wx_fmt=png)]

3

发现 enum 类型的序列化器是 EnumSerializer， 看看 EnumSerializer 的 serde 实现，如图所示：

4

最关键的两个变量：

1. 序列化时用 valueToOrdinal

2. 反序列化时用 values

从而印证了上面的说法。flink enum 序列化时使用的是枚举值下标进行 serde，因此一旦枚举值顺序发生改变，或者添加、删除一个枚举值，就会导致其他枚举值的下标出现错位的情况。从而导致数据错误。

5.避坑篇-如何避免这种问题

5.1.枚举解决

在上述场景中，如果又想要把新枚举值加进去，又需要状态能够正常恢复，正常产出数据。

那么可以把新的枚举值在尾部添加，比如下面这样。

enum DimNameEnum {province,age,sex,uv_type, // 添加在尾部;
}

5.2.非枚举解决

还有一种方法如标题，就是别用枚举值，直接用 string 就 vans 了。

6.总结篇

本文主要介绍了 flink 枚举值 serde 中的坑，当在 enum 中添加删除枚举值时，就有可能导致状态岔劈。随后给出了原因是由于 enum serde 器的实现导致的这种情况，最后给出了解决方案。

大数据羊说

用数据提升美好事物发生的概率~

30篇原创内容

公众号

往期推荐

[

flink sql 知其所以然（四）| sql api 类型系统

](http://mp.weixin.qq.com/s?__biz=MzkxNjA1MzM5OQ==&mid=2247488788&idx=1&sn=0127fd4037788762a0401313b43b0ea5&chksm=c15499ecf62310fa747c530f722e631570a1b0469af2a693e9f48d3a660aa2c15e610653fe8c&scene=21#wechat_redirect)

[

flink sql 知其所以然（三）| 自定义 redis 数据汇表（附源码）

](http://mp.weixin.qq.com/s?__biz=MzkxNjA1MzM5OQ==&mid=2247488720&idx=1&sn=5695e3691b55a7e40814d0e455dbe92a&chksm=c1549828f623113e9959a382f98dc9033997dd4bdcb127f9fb2fbea046545b527233d4c3510e&scene=21#wechat_redirect)

[

flink sql 知其所以然（二）| 自定义 redis 数据维表（附源码）

](http://mp.weixin.qq.com/s?__biz=MzkxNjA1MzM5OQ==&mid=2247488635&idx=1&sn=41817a078ef456fb036e94072b2383ff&chksm=c1549883f623119559c47047c6d2a9540531e0e6f0b58b155ef9da17e37e32a9c486fe50f8e3&scene=21#wechat_redirect)

[

flink sql 知其所以然（一）| source\sink 原理

](http://mp.weixin.qq.com/s?__biz=MzkxNjA1MzM5OQ==&mid=2247488486&idx=1&sn=b9bdb56e44631145c8cc6354a093e7c0&chksm=c1549f1ef623160834e3c5661c155ec421699fc18c57f2c63ba14d33bab1d37c5930fdce016b&scene=21#wechat_redirect)

[

揭秘字节跳动埋点数据实时动态处理引擎（附源码）

](http://mp.weixin.qq.com/s?__biz=MzkxNjA1MzM5OQ==&mid=2247488435&idx=1&sn=5d89a0d24603c08af4be342462409230&chksm=c1549f4bf623165d977426d13a0bdbe821ec8738744d2274613a7ad92dec0256d090aea4b815&scene=21#wechat_redirect)

更多 Flink 实时大数据分析相关技术博文，视频。后台回复 “flink” 获取。

点个赞+在看，感谢您的肯定 ?