flink-user-zh mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From 戴嘉诚 <a773807...@gmail.com>
Subject Re: Re: Flink checkpoint 并发问题
Date Thu, 25 Jul 2019 14:39:44 GMT
嗯嗯!感谢

Yun Tang <myasuka@live.com> 于2019年7月25日周四 下午9:37写道:

> Hi
>
> 你给的代码跟你的异常栈其实还是对不上,前文已经说了,出问题的是operator
state,但是你的代码都是keyed
> state相关的代码。不过从你出问题的operator name "KeyedProcess -> async
wait operator ->
> Flat Map -> Sink",
> 以及异常栈中的StreamElementSerializer使用和一致性问题的表象,我推测应该是应该是AsyncWaitOperator中的operator
> state "_async_wait_operator_state_"相关。最近fix的
> https://issues.apache.org/jira/browse/FLINK-13063
> 问题,正是暂时解决一致性问题,可以考虑cherry pick相关的fix重新部署你们的Flink作业,观察该问题是否还会复现。
>
> 祝好
> 唐云
> ________________________________
> From: 戴嘉诚 <a773807943@gmail.com>
> Sent: Thursday, July 25, 2019 21:07
> To: user-zh <user-zh@flink.apache.org>
> Subject: Re: Re: Flink checkpoint 并发问题
>
> Hi 唐云
>
>
> 这个问题在这个job上是必现的,即使是failvoer后自动重启,也会出现,但是在同一个处理逻辑,在其他的场景下是没有出现过,而这其中的区别是,这个job是数据量比较大,将近一分钟80万左右。
>
> 这个其中的ProcessElement大致的代码是这样的:
> .process(new KeyedProcessFunction<String, Map<String, Object>, Map<String,
> Object>>() {
>           private static final long serialVersionUID =
> 5245290869789704294L;
>
>           private MapState<String, Long> accumulateStateMap;
>           Map<String, Object> resultMap = new HashMap<>();
>           private transient Long hourClear = 24L;
>
>
>           @Override
>           public void open(Configuration parameters) throws Exception {
>             MapStateDescriptor<String, Long> accumulateState = new
> MapStateDescriptor<>("accumulateState", StringSerializer.INSTANCE,
> LongSerializer.INSTANCE);
>             accumulateStateMap =
> getRuntimeContext().getMapState(accumulateState);
>           }
>
>           @Override
>           public void processElement(Map<String, Object> value, Context
> ctx,
>               Collector<Map<String, Object>> out) throws Exception {
>             logger.info("来数据了:{}", value);
>             realData.increment();
>             resultMap.clear();
>             String valueFieldValue =
> String.valueOf(value.get(getLastName(valueFieldName)));
>             Long timeFieldValue =
> Long.parseLong(String.valueOf(value.get(timeFieldName)));
>             //写到state中
>             //判断,state是否存在fieldValue,  如果fieldValue
> 存在,再判断state的时间是否小于time(用于判断乱序时间
>             if (!accumulateStateMap.contains(valueFieldValue) ||
> accumulateStateMap.get(valueFieldValue) < timeFieldValue) {
>               accumulateStateMap.put(valueFieldValue, timeFieldValue);
>             }
>             //判断配置是否已经刷进来了
>             if (value.containsKey("config")) {
>               Map<String, String> config = (Map<String, String>)
> value.get("config");
>               Integer configCount =
> Integer.parseInt(config.get(countFieldName));
>               Long configTime =
> Long.parseLong(config.get(timeRangeFieldName)) * 1000;
>               //在配置时间范围前
>               long lastTimeStamp = timeFieldValue - configTime;
>               //状态里面有多少个值
>               int stateSize = 0;
>               //遍历state, 删除过时的时间
>               Iterator<Entry<String, Long>> iterator =
> accumulateStateMap.iterator();
>               while (iterator.hasNext()) {
>                 ++stateSize;
>                 Entry<String, Long> next = iterator.next();
>                 if (lastTimeStamp >= next.getValue()) {
>                   iterator.remove();
>                   --stateSize;
>                 }
>               }
>               //state的值的数量大于阈值
>               if (stateSize >= configCount) {
>                 resultMap.put("id", config.get("id"));
>                 resultMap.put("config_id", config.get("_id"));
>                 resultMap.put("config_version", config.get("_version"));
>                 resultMap.put("config_score", config.get("score"));
>                 resultMap.put("config_ttl", config.get("ttl"));
>                 resultMap.put("startTime", lastTimeStamp);
>                 resultMap.put("endTime", timeFieldValue);
>                 resultMap.put("key", ctx.getCurrentKey());
>                 resultMap.put("value", valueFieldValue);
>                 resultMap.put("count", stateSize);
>                 out.collect(resultMap);
>               }
>               logger.info("当前key为:{}, 聚集数量为:{}", ctx.getCurrentKey(),
> stateSize);
>               //根据配置时间乘以2,错信息范围加上来注册指定清理state的时间
>               hourClear = (configTime * 2  + EXPIRATION_TIME) / 3600000;
>               LocalDateTime localDateTime =
>
> Instant.ofEpochMilli(timeFieldValue).atZone(ZoneId.systemDefault()).toLocalDateTime();
>               localDateTime =
> localDateTime.withMinute(0).withSecond(0).withNano(0).plusHours(hourClear);
>               long timeClean =
> localDateTime.toInstant(OffsetDateTime.now().getOffset()).toEpochMilli();
>               timeTimerGauge = String.valueOf(timeClean);
>               //注册过期时间
>               ctx.timerService().registerEventTimeTimer(timeClean);
>             }
>           }
>
>           @Override
>           public void onTimer(long timestamp, OnTimerContext ctx,
>               Collector<Map<String, Object>> out) throws Exception {
>             //减去时间,防止删除中间要累积的数据
>             Long lasttime = timestamp - (hourClear * 3600000);
>             //删除过期时间
>             int stateSize = 0;
>             int removeState = 0;
>             Iterator<Entry<String, Long>> iterator =
> accumulateStateMap.iterator();
>             while (iterator.hasNext()) {
>               ++stateSize;
>               Entry<String, Long> next = iterator.next();
>               if (lasttime >= next.getValue()) {
>                 iterator.remove();
>                 --stateSize;
>                 ++removeState;
>               }
>             }
>             if (stateSize == 0) {
>               accumulateStateMap.clear();
>             }
>             //把这个定时器删除掉
>             ctx.timerService().deleteEventTimeTimer(timestamp);
>           }
>         })
>
> Yun Tang <myasuka@live.com> 于2019年7月25日周四 下午8:39写道:
>
> > Hi 戴嘉诚
> >
> > 从异常栈上看,DefaultOperatorStateBackendSnapshotStrategy 88行左右的代码[1]是对list
> > state进行deepcopy,但是你告诉我的state descriptor却是MapStateDescriptor<String,
> >
> Long>,这二者之间明显是对不上的,还请再校验一下你的代码。异常栈出现在checkpoint同步阶段的内容拷贝时的ConcurrentModification,这个确实是很奇怪的,所以还需要麻烦回答一下下面几个问题:
> >
> >   *   这个问题是必现的么?作业failover或者重新提交之后也会出现么?
> >   *   由于operator
> >
> state的snapshot/restore需要由用户提供,所以如果你继承了ListCheckpointed接口,请提供一下snapshotState和restoreState接口的实现(如果是继承了CheckpointedFunction接口,请提供一下snapshotState和initializeState的实现)。另外,这个operator
> > state的申明以及相关的使用地方也最好提供一下。
> >
> > [1]
> >
> https://github.com/apache/flink/blob/480875f045a9777877ed1a90f9e0c6e01b7e03c9/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java#L88
> >
> > 祝好
> > 唐云
> > ________________________________
> > From: 戴嘉诚 <a773807943@gmail.com>
> > Sent: Thursday, July 25, 2019 19:26
> > To: user-zh <user-zh@flink.apache.org>
> > Subject: Re: Re: Flink checkpoint 并发问题
> >
> > hi
> >     你好,我用的flink是1.8,但是是根据hadoop 2.7.3.2.6.5.0-292  自行打包的,
 operator state
> > descriptor是使用MapStateDescriptor<String, Long>,
> > 谢谢!
> >
> > Yun Tang <myasuka@live.com> 于2019年7月25日周四 下午7:10写道:
> >
> > > Hi  all
> > >
> > > 你们讨论的已经越来越偏了,出问题的是operator state
> > > backend,实际上与RocksDB没有关系,与MaxConcurrentCheckpoints 也不应该有关系。
> > >
> > > To 戴嘉诚
> > > 你使用的Flink版本是什么?这个operator 里面的operator state descriptor是什么?
> > >
> > > 祝好
> > > 唐云
> > > ________________________________
> > > From: 戴嘉诚 <a773807943@gmail.com>
> > > Sent: Thursday, July 25, 2019 19:04
> > > To: user-zh@flink.apache.org <user-zh@flink.apache.org>
> > > Subject: Re: Re: Flink checkpoint 并发问题
> > >
> > > 这个用window不符合这个场景,因为是取每时每刻的最近半小时,而window只能是固定时间区间,例如window只能统计6点半至7点,
> > > 7点之7点半,而我的场景是需要6点到6点半,6点02分到6点32分,是根据数据的时间来确定区间,达到统计区间内的总数
> > >
> > > athlon512@gmail.com <athlon512@gmail.com>于2019年7月25日 周四18:50写道:
> > >
> > > > 那你用window和evictor 不可以吗?
> > > > 类似这样,因为我理解你的业务需求可以用这个来实现
> > > >
> > 在flink源码中org.apache.flink.streaming.examples.windowing.TopSpeedWindowing
> > > >
> > > > ------------------------------
> > > > athlon512@gmail.com
> > > >
> > > >
> > > > *发件人:* 戴嘉诚 <a773807943@gmail.com>
> > > > *发送时间:* 2019-07-25 18:45
> > > > *收件人:* user-zh <user-zh@flink.apache.org>
> > > > *主题:* Re: Re: Flink checkpoint 并发问题
> > > >
> > > >
> > > >
> > >
> >
> 这个应该不行,因为这个ttl是删除整个map状态,而我是需要最近半小时内的数据量,所以这个map状态,如果一直有数据写入,我就只需要获取最近的半小时,超过的就在map里面删除,这个ttl不能精确到map里面的具体的k-v
> > > > 对
> > > >
> > > > athlon512@gmail.com <athlon512@gmail.com> 于2019年7月25日周四
下午6:40写道:
> > > >
> > > > > 其实你可以不用自己删除.使用TTL设置短一些时间,试试
> > > > >
> > > > >
> > > > >
> > > > > athlon512@gmail.com
> > > > >
> > > > > 发件人: 戴嘉诚
> > > > > 发送时间: 2019-07-25 18:24
> > > > > 收件人: user-zh
> > > > > 主题: Re: Flink checkpoint 并发问题
> > > > > 你好,这个参数我没有设置过,对于checkpoint,我只是设置了rockDB增量写入,其他都没有设置了
> > > > >
> > > > > athlon512@gmail.com <athlon512@gmail.com> 于2019年7月25日周四
下午6:20写道:
> > > > >
> > > > > > setMaxConcurrentCheckpoints 这个参数你设置过么?
> > > > > >
> > > > > >
> > > > > >
> > > > > > athlon512@gmail.com
> > > > > >
> > > > > > 发件人: 戴嘉诚
> > > > > > 发送时间: 2019-07-25 18:07
> > > > > > 收件人: user-zh
> > > > > > 主题: Flink checkpoint 并发问题
> > > > > > 大家好:
> > > > > >
> > > > > >     我这里有个Job,会经常不定时的挂掉重启,具体的异常日志为一下,大致是checkpoint的同步失败的
> > > > > >
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> 在这个job中,具体是查询最近半小时的数据量是否达到配置的阈值,我使用了rockDB作为后端,然后中间会有个processElement方法会把Key出来的消息,在内部中的MapState堆积,同时也会每条消息过来,就遍历一遍MapState,来删除其中过时的数据。
> > > > > >
> > > > > >
> > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
> 这个异常是不是我在ProcessElement方法内对MapState进行删除的过程中,checkpoint异步复制这个state到rockDB导致的?
> > > > > >
> > > > > >
> > > > > > java.lang.Exception: Could not perform checkpoint 550 for
> operator
> > > > > > KeyedProcess -> async wait operator -> Flat Map ->
Sink:
> 写入redis库存
> > > > > > (16/20).
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:595)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.io.BarrierBuffer.notifyCheckpoint(BarrierBuffer.java:396)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.io.BarrierBuffer.processBarrier(BarrierBuffer.java:292)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.io.BarrierBuffer.getNextNonBlocked(BarrierBuffer.java:200)
> > > > > >
> > > > > >          at org.apache.flink.streaming.runtime.io
> > > > > > .StreamInputProcessor.processInput(StreamInputProcessor.java:209)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:105)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
> > > > > >
> > > > > >          at
> > > > org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
> > > > > >
> > > > > >          at java.lang.Thread.run(Thread.java:748)
> > > > > >
> > > > > > Caused by: java.lang.Exception: Could not complete snapshot
550
> for
> > > > > > operator KeyedProcess -> async wait operator -> Flat Map
-> Sink:
> > > > > > 写入redis库存 (16/20).
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:422)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1113)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1055)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:729)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:641)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:586)
> > > > > >
> > > > > >          ... 8 more
> > > > > >
> > > > > > Caused by: java.util.ConcurrentModificationException
> > > > > >
> > > > > >          at
> > > java.util.HashMap$HashIterator.nextNode(HashMap.java:1442)
> > > > > >
> > > > > >          at
> java.util.HashMap$EntryIterator.next(HashMap.java:1476)
> > > > > >
> > > > > >          at
> java.util.HashMap$EntryIterator.next(HashMap.java:1474)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> com.esotericsoftware.kryo.serializers.MapSerializer.copy(MapSerializer.java:156)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> com.esotericsoftware.kryo.serializers.MapSerializer.copy(MapSerializer.java:21)
> > > > > >
> > > > > >          at com.esotericsoftware.kryo.Kryo.copy(Kryo.java:862)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:248)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.copy(StreamElementSerializer.java:105)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.runtime.streamrecord.StreamElementSerializer.copy(StreamElementSerializer.java:46)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.state.ArrayListSerializer.copy(ArrayListSerializer.java:73)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.state.PartitionableListState.<init>(PartitionableListState.java:68)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.state.PartitionableListState.deepCopy(PartitionableListState.java:80)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.state.DefaultOperatorStateBackendSnapshotStrategy.snapshot(DefaultOperatorStateBackendSnapshotStrategy.java:88)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.runtime.state.DefaultOperatorStateBackend.snapshot(DefaultOperatorStateBackend.java:261)
> > > > > >
> > > > > >          at
> > > > > >
> > > > >
> > > >
> > >
> >
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:402)
> > > > > >
> > > > > >          ... 13 more
> > > > > >
> > > > >
> > > >
> > > >
> > >
> >
>
Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message