Hi
可以的话也同步下相关的计算逻辑，从 checkpoint 恢复后的统计结果可能会和计算逻辑有关
Best,
Congxian
Hangxiang Yu 于2022年10月10日周一 14:04写道：
> 是什么值下跌呢？哪个metric吗？
>
> On Mon, Oct 10, 2022 at 1:34 PM 天下五帝东 wrote:
>
> > Hi:
> > 各位大佬们，flink任务从状态恢复时，统计值有一个明显的下跌，等ck打完后，统计值才会上升，这种情况有什么好的解决方案吗？
>
>
>
> --
> Best,
> Hangxiang.
Hi
有 savepoint/checkpoint 失败时的具体 jobmanager log 以及失败 task 对应的 taskmanager log
的话可以发一下，大家帮助看一下
Best,
Congxian
Xuyang 于2022年8月30日周二 23:18写道：
>
> Hi，看起来这个报错是用于输出信息的文件找不到了，可以尝试加一下这个配置再试一下“taskmanager.log.path”，找一下导致tasks超时的根本原因。
> 还可以试一下用火焰图或jstack查看一下那几个tasks超时的时候是卡在哪个方法上。
>
>
>
>
>
>
>
>
>
>
>
从日志看 checkpoint 超时了，可以尝试看一下是哪个算子的哪个并发没有做完 checkpoint，可以看看这篇文章[1] 能否帮助你
[1] https://www.infoq.cn/article/g8ylv3i2akmmzgccz8ku
Best,
Congxian
Frost Wong 于2021年3月18日周四 下午12:28写道：
> 哦哦，我看到了有个
>
> setTolerableCheckpointFailureNumber
>
> 之前不知道有这个方法，倒是可以试一下，不过我就是不太理解为什么会失败，也没有任何报错
>
Hi
从你的日志看作业启动失败的原因是：
Caused by: java.lang.IllegalArgumentException: Wrong FS:
hdfs://xx/flink120/, expected: file:///
看上去你设置的地址和 需要的 schema 不一样，你需要解决一下这个问题
Best,
Congxian
todd 于2021年3月15日周一 下午2:22写道：
> 通过脚本提交flink作业,提交命令：
> /bin/flink run-application -t yarn-application
>
对于 keyed state，需要保证同一个 key 在 同一个 keygroup 中，如果是某个 key 有热点，可以在 keyby 之前进行一次
map（在 key 后面拼接一些 后缀），然后 keyby，最后处理完成之后，将这些进行聚合
Best,
Congxian
guomuhua <663021...@qq.com> 于2021年3月4日周四 下午12:49写道：
> 我也遇到类似情况，为了打散数据，keyby加了随机数。请问怎么正确打散数据呢？
> nobleyd wrote
> > 是不是使用了随机key。
>
> > guaishushu1103@
>
> >
>
Hi
你 flink 是什么版本，以及你作业 checkpoint/state 相关的配置是什么呢？如果可以的话，把完整的 jm log 发一下
Best,
Congxian
chen310 <1...@163.com> 于2021年2月1日周一 下午5:41写道：
> 补充下，jobmanager日志异常：
>
> 2021-02-01 08:54:43,639 ERROR
> org.apache.flink.runtime.rest.handler.job.JobDetailsHandler [] - Exception
> occurred in REST
Hi
你可以看一下是不是 作业的状态变成了 FINISH 了，现在 Flink 在 FINISH 的时候，会删除 checkpoint（就算设置了
retain on cancel 也会删除）
PS ：FLINK-18263 正在讨论是否在 Job 变为 FINISH 状态后保留 checkpoint
[1] https://issues.apache.org/jira/browse/FLINK-18263
Best,
Congxian
yinghua...@163.com 于2021年1月15日周五 上午11:23写道：
>
Hi
state.backend 应该是你在 flink-conf 中设置了这个值。具体到你这里的情况，最终的配置是
RocksDB（以代码为准，如果代码没有设置会使用 flink-conf 中的文件）。你可以看看 TM 日志，应该可以看到更详细的信息
Best,
Congxian
bradyMk 于2020年12月15日周二 下午5:05写道：
> Hi，想请教大家一个问题，我用单任务配置使用rocksDB状态后端，代码如下：
>
> val backend = new RocksDBStateBackend(path, true)
>
可以看一下 remote task 对应的 tm 日志，看看有没有啥异常
Best,
Congxian
赵一旦 于2020年12月2日周三 下午6:17写道：
> 我都是80G、100G这么分配资源的。。。
>
> guanxianchun 于2020年10月28日周三 下午5:02写道：
>
> > flink版本： flink-1.11
> > taskmanager memory: 8G
> > jobmanager memory: 2G
> > akka.ask.timeout：20s
> > akka.retry-gate-closed-for： 5000
> >
checkpoint 失败了可以看看 是超时了，还是有 task snapshot 失败了，可以从 JM log
中来发现。超时的话，可以看下是数据量大需要时间久，还是 timeout 啥的设置太短；异常的话可以从对应的 tm log 看下为啥 snapshot
失败了
Best,
Congxian
王默 于2020年11月27日周五 下午11:43写道：
> checkpoint失败是在web页面上发现的，您看下截图https://imgchr.com/i/Dr3PNn
> 看taskmanager日志确实没有超时，也没有其他异常
>
>
>
>
>
>
>
>
>
>
>
>
Hi
退订请发邮件到 user-zh-unsubscr...@flink.apache.org，详情可以参考文档[1]
[1]
https://flink.apache.org/community.html#how-to-subscribe-to-a-mailing-list
Best,
Congxian
回响 <939833...@qq.com> 于2020年11月24日周二 下午8:42写道：
>
Hi
异常信息中有 “Failed to trigger savepoint. Failure reason: An Exception
occurred while triggering the checkpoint.” 或许你可以看看 JM 的日志，找一下看看有没有什么详细日志
Best,
Congxian
张锴 于2020年11月7日周六 下午4:14写道：
> 本人用flink 1.10.1版本进行savepoint时遇到下列错误，暂时不清楚错误的原因，特来寻求帮助，麻烦大佬们看看
>
>
hi
退订请发邮件到 user-zh-unsubscr...@flink.apache.org，更多详情请参考[1]
[1] https://flink.apache.org/community.html#mailing-lists
Best,
Congxian
李郝 <13777597...@163.com> 于2020年11月5日周四 下午9:54写道：
> 退订
Hi
从 client 端日志，或者 JM 日志还能看到其他的异常么？
Best,
Congxian
张锴 于2020年11月6日周五 上午11:42写道：
> 重启和反压都正常
> 另外增加了从客户端到master的时间，还是有这个问题
>
> hailongwang <18868816...@163.com> 于 2020年11月6日周五 10:54写道：
>
> > Hi,
> >
> >
> > 这个报错只是在规定的时间内没有完成 Savepoint，导致客户端连接 Master 超时，
> > 具体的原因需要看下 Jobmaster 的日志。
> >
Hi
这个问题看上去是特定 JDK 版本上，某些写法下对象被提前回收了，猜测和 gc 有关。之前看到一个可能相关的帖子[1]
[1] https://cloud.tencent.com/developer/news/564780
Best,
Congxian
蒋佳成(Jiacheng Jiang) <920334...@qq.com> 于2020年11月4日周三 上午11:33写道：
> hi，这个问题我也遇到了，这个问题的根本原因是啥呢？
>
>
>
> --原始邮件--
> 发件人:
Hi
你可以参考一下这个文档[1] 以及文档内的链接，这里面有些如何贡献相关的内容
[1] https://flink.apache.org/contributing/how-to-contribute.html
Best,
Congxian
Xingbo Huang 于2020年11月3日周二 上午9:29写道：
> Hi,
> 需要committer才有权限进行assign的，你可以在JIRA下面ping一下对应模块的committer
> Best,
> Xingbo
>
> zihaodeng <284616...@qq.com> 于2020年11月3日周二
Hi
我理解你的 BroadcastStream 也会定期的再次读取，然后更新对应的状态，这样的话，你的 source 可以一直在读取数据（run
函数），不退出即可。如果只希望读取一次话，是不是维表也可以满足你的需求呢？
Best,
Congxian
restart 于2020年11月3日周二 上午10:11写道：
> 问题：job在接入广播流后，checkpint失效。
>
>
Hi
退订请发邮箱到 user-zh-unsubscr...@flink.apache.org 更详细的可以参考这个文档[1]
[1] https://flink.apache.org/community.html#mailing-lists
Best,
Congxian
196371551 <196371...@qq.com> 于2020年11月3日周二 下午2:08写道：
> 退订
sm（ parallelism
> ） 这种方式好多了。
>
>
> 请问下后续会扩展keyby接口不？keyby 可以根据自定义partition，然后返回keyedstream。
>
>
> Best Wishes.
>
>
>
> Congxian Qiu 于2020年11月2日周一 下午1:52写道：
>
> > Hi
> > 不太明白你这里问题，如果是说按照 Md5 进行 keyby 不均匀，直接用 hashcode keyby 会更均匀的话，是不是直接把计算
> >
Hi
不太明白你这里问题，如果是说按照 Md5 进行 keyby 不均匀，直接用 hashcode keyby 会更均匀的话，是不是直接把计算
md5 的逻辑改成计算 hashcode 的逻辑就行了
Best,
Congxian
Peihui He 于2020年11月2日周一 上午10:01写道：
> hi，
>
> 已经尝试过了，通过partitionCustom 返回的是datastream，如果后面接cep的化，并发度就是1了，性能更不上。
>
> Best Wishes.
>
> Zhang Yuxiao 于2020年10月31日周六 上午9:38写道：
>
>
Hi
如果是空间满，可以看看都是什么文件（什么目录，文件名是否有某些格式等），这些文件的存在是否合理，如果你觉得不合理可以发到社区邮件列表讨论，或者创建
issue 进行跟进。如果是合理的那就只能想办法删除一些不需要的文件了
Best,
Congxian
赵一旦 于2020年10月30日周五 下午5:51写道：
> 磁盘满看是什么东西导致满，然后清理就是了。比如是flink日志满？那就清理flink日志。
>
> void <2030531...@qq.com> 于2020年10月29日周四 下午6:56写道：
>
> > hi all
> >
Hi
1 Flink 的 RestartStrategy[1] 可以解决你的问题吗？
2 从 checkpoint 恢复 这个，可以尝试记录每个作业最新的 checkpoint 地址，也可以在启动的时候从 hdfs 获取一下
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/task_failure_recovery.html
Best,
Congxian
bradyMk 于2020年10月30日周五 上午11:51写道：
>
>
Hi
这个你可以尝试把这个信息记录到哪里，或者在启动的时候从这个 jobId 的目录下去查找所有的 chk-xxx 然后选择一个合适的 目录进行恢复
Best,
Congxian
marble.zh...@coinflex.com.INVALID
于2020年10月27日周二 下午4:54写道：
> 刚钉钉群里建议我把路径指到jobId/chk-xx目录，这样就可以恢复了。
>
> 但是如果这样，这个xx随着checkpoint的变化而变化，这样怎么做到自动提交job?
>
>
>
> --
> Sent from:
Hi
从报错看，你知道的是一个目录，这个目录下面没有 _metadata 文件，这不是一个完整的 checkpoint/savepoint
因此不能用于恢复
Best,
Congxian
marble.zh...@coinflex.com.INVALID
于2020年10月27日周二 下午4:06写道：
> /opt/flink/bin/flink run -d -s /opt/flink/savepoints -c
> com.xxx.flink.ohlc.kafka.OrderTickCandleView
>
Hi smq
这两个东西不相互影响，理论上 checkpoint 允许失败次数这个只会导致 job fail，而重启策略则是在 job fail
的时候判断怎么继续，如果不符合预期，可以看一下 jm 的 log 或者分享一下 jm log 让大家帮忙看看
Best,
Congxian
smq <374060...@qq.com> 于2020年10月27日周二 上午11:25写道：
> 各位大佬好：
> 我现在设置容忍checkpoint失败次数是0,重启策略为固定延时重启，重启100次。
>
Hi
从报错来看是无法在 "/opt/flink/savepoints" 这个路径下创建目录，这个错误下面应该应该一些异常信息，可以看下具体是啥原因。
另外，使用本地路径存储 checkpoint/savepoint 的话，那么恢复的时候，需要确保该文件能被新的 JM/TM 所访问到
Best,
Congxian
marble.zh...@coinflex.com.INVALID
于2020年10月26日周一 下午3:32写道：
> 你好，
> 我在flink jobmanager里的flink-conf.yaml添加了以加三个个关的state配置参数，
>
Hi
退订请发邮件到 user-zh-unsubscr...@flink.apache.org 更多详细情况可以参考[1]
[1] https://flink.apache.org/community.html#mailing-lists
Best,
Congxian
费文杰 <15171440...@163.com> 于2020年10月20日周二 下午1:51写道：
> hi：
> 退订！
Hi
你可以看下 JM log 中这个 savepoint 失败是什么原因导致的，如果是 savepoint 超时了，就要看哪个 task
完成的慢，（savepoint 可能比 checkpoint 要慢）
Best,
Congxian
Robin Zhang 于2020年10月19日周一 下午3:42写道：
> 普通的source -> map -> filter-> sink 测试应用。
>
> 触发savepoint的脚本 ：
> ${FLINK_HOME} stop -p ${TARGET_DIR} -d ${JOB_ID}
> 具体报错信息：
>
>
Hi
或许你可以使用 timer 来进行兜底，注册一个未来某个时间的 timer，然后 timer 触发的时候把 state 清理掉
Best,
Congxian
x <35907...@qq.com> 于2020年10月19日周一 下午2:55写道：
> 版本为v1.10.1
>
Hi
To subscribe to the user mail list, you need to send a mail to
user-zh-subscr...@flink.apache.org, you can get more info here[1]
可以发送邮件到 user-zh-subscr...@flink.apache.org 订阅 user-zh 邮件列表
https://flink.apache.org/community.html#mailing-lists
Best,
Congxian
fangzhou ding
FYI 分享一个可能相关的文章[1]
[1] https://cloud.tencent.com/developer/news/564780
Best,
Congxian
Storm☀️ 于2020年10月15日周四 上午10:42写道：
> 非常感谢。
> 后续我关注下这个问题，有结论反馈给大家，供参考。
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
Hi
容易异常退出是指 container 退出吗？可以看下 JM/TM log 是否有相应信息，如果没有，可以尝试从 yarn 侧看下日志为什么
container 退出了
Best,
Congxian
caozhen 于2020年10月12日周一 下午6:08写道：
>
> 可以发下 "分配完applicationid后，容器经常异常退出" 产生的错误日志吗？
>
> 或者排查下flink客户端中的错误日志，以及yarn-historyserver里的日志。
>
>
>
> Dream-底限 wrote
> > hi
> >
Hi
从错误日志看，应该是 filesystem 相关的配置（或者 jar 包）有问题，可以参考下这个邮件列表[1]看看能否解决你的问题
[1]
http://apache-flink.147419.n8.nabble.com/Flink-1-11-1-on-k8s-hadoop-td5779.html#a5834
Best,
Congxian
superainbower 于2020年9月30日周三 下午3:04写道：
> 补充一下，我的错误日志
> Caused by:
Hi, @Storm 请问你用的是 flink 是哪个版本，然后栈是什么呢？可以把相关性信息回复到这里，可以一起看看是啥问题
Best,
Congxian
大森林 于2020年10月10日周六 下午1:05写道：
> 我这边是老版本的jdk8，和jdk261没啥关系的
>
>
>
>
> --原始邮件--
> 发件人:
> "user-zh"
>
Hi
增量 checkpoint 是指，每次只上传的 *必须的* sst 文件。因为 RocksDB 生成的 sst 文件是不可变的，所以之前上传过的
sst 文件直接引用即可，这样减少了很多 sst 文件的上传（也减少了 HDFS 的存储和删除等操作）
Best,
Congxian
宁吉浩 于2020年10月9日周五 上午10:20写道：
> 没看过源码，看过一些文档，结论还需验证(应该不用了)。
> 增量checkpoint指的是
>
Hi
RocksDB 里面存的是 State，Flink 在做 checkpoint 的时候会把 State 备份到 HDFS 上，如果失败的话从
Checkpoint 进行恢复，如果想了解更详细的内容，可以参考文档[1][2] 以及文档里面的链接
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/state/checkpointing.html
[2]
Hi
checkpoint 会按照用户设定的周期定期触发, 同时也会收到 minPauseBetweenCheckpoints 以及
maxConcurrentCheckpoints 等参数的控制，具体的可以看一下这个文档[1]，
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/state/checkpointing.html#enabling-and-configuring-checkpointing
Best,
Congxian
郝文强
Hi Eleanore
What the `CheckpointRetentionPolicy`[1] did you set for your job? if
`ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION` is set, then the
checkpoint will be kept when canceling a job.
PS the image did not show
[1]
Hi
这个问题是应该和 FLINK-17479 是一样的，是特定 JDK 上会遇到问题，可以考虑升级一下 flink 版本，或者替换一个 JDK 版本
Best,
Congxian
Storm☀️ 于2020年9月27日周日 上午10:17写道：
> 各位好，checkpoint相关问题L
>
> flink版本1.10.1:，个别的checkpoint过程发生问题：
> java.lang.Exception: Could not perform checkpoint 1194 for operator Map
> (3/3).
> at
>
>
Hi
退订请发邮件到 user-zh-unsubscr...@flink.apache.org
详情可以参考文档[1]
[1] https://flink.apache.org/community.html#mailing-lists
Best,
Congxian
Han Xiao(联通集团联通支付有限公司总部) 于2020年9月18日周五 下午2:52写道：
>
> 退订
>
> 如果您错误接收了该邮件，请通过电子邮件立即通知我们。请回复邮件到
> hqs-s...@chinaunicom.cn，即可以退订此邮件。我们将立即将您的信息从我们的发送目录中删除。
>
Thanks for being the release manager Zhu Zhu and everyone involved in!
Best,
Congxian
Weijie Guo 2 于2020年9月18日周五 下午11:42写道：
> Good job! Very thanks @ZhuZhu for driving this and thanks for all
> contributed
> to the release!
>
> best,
> Weijie
> Zhu Zhu-2 wrote
> > The Apache Flink community
Hi
没有太明白你这里为什么数据较少的 source 会覆盖数据更多的 source。能否详细描述下呢？
如果是因为某些 source 的数据把 watermark 给推高了，从而导致其他的 source
数据变成迟到的数据，那么我理解这个暂时没法直接解决的，因为 watermark 是 task 级别的。如果你想要在一个 Flink
作业中实现更细粒度的 watermark，或许你可以尝试下 processfuncton[1] 自己来定制一套类似的机制
另外你也可以看下这个文档[2] 看是否在你的场景中有帮助
[1]
Hi
这个需求可以考虑用 processfunction[1] 来实现，window 的 state 只能给单独的 window 使用，其他
window 不能操作之前的 window 的 state
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/operators/process_function.html#the-keyedprocessfunction
Best,
Congxian
marble.zh...@coinflex.com.INVALID
Hi
如果你们删除过文件的话，那么你可以检查 chk-xxx 目录下是否有 _metadata 文件存在，存在基本就是可用的
checkpoint。如果你有删除过文件的话，则需要读取 _metadata 文件，然后看看是否所有文件都存在。
Best,
Congxian
Dream-底限 于2020年9月16日周三 上午10:27写道：
> hi、
> 我正在做基于检查点的任务自动恢复，请问有没有什么方法来检查检查点是否是一个完整有效的检查点，因为有的时候检查点会失败，所以有没有api层面的校验方式
>
at
> > >
> >
> org.apache.flink.table.client.gateway.local.ExecutionContext.createStreamExecutionEnvironment(ExecutionContext.java:691)
> > > at
> > >
> >
> org.apache.flink.table.client.gateway.local.ExecutionContext.createTableEnvironment(Execution
Hi
如果你的 state 都非常小的话，可能就会保存在 meta 文件中了，这样的话就只有 _metadata
这一个文件的。具体逻辑可以看一下这里[1]
[1]
https://github.com/apache/flink/blob/9b0fb562898b809b860cf0065ded7a45c49300af/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/metadata/MetadataV2V3SerializerBase.java#L442
Best,
Congxian
Hi
据我所知，iceberg 有一个 flink 的 sink，可以看下这个 PR[1]
[1] https://github.com/apache/iceberg/pull/856
Best,
Congxian
dixingxing85 于2020年9月12日周六 下午4:54写道：
> Hi all：
> 想请教一个问题，现在一些公司已经开始应用数据湖技术了，目前flink和iceberg,hudi,delta
> lake这些的集成情况分别是怎样的？社区有主动跟进某一个数据湖技术的集成吗？预计什么时候能有相对完善的source,sink。谢谢
>
> Sent
Hi
index 相关的，或许可以看下这个文档[1]
[1]
https://github.com/facebook/rocksdb/wiki/Block-Cache#caching-index-filter-and-compression-dictionary-blocks
Best,
Congxian
范瑞 <836961...@qq.com> 于2020年9月2日周三 下午2:05写道：
> 再补充一下正常 subtask 火焰图：
>
Hi
1 你的作业运行的是哪个版本
2 你作业挂掉应该是 tolerable failure threshold 超了导致的，这个可以在 checkpoint config
中进行配置，这样 checkpoint 失败后不会导致作业失败
3 如果可以的话，你可以上传一下 jm 和 tm log
Best,
Congxian
jordan95225 于2020年9月7日周一 上午11:05写道：
> Hi，
> 我现在有一个flink任务，运行一段时间后checkpoint会超时，INFO信息如下：
> checkpoint xxx of job xxx
Hi
对于 checkpoint size 持续变大的情况，可以考虑下：
1）你使用啥 backend，是否使用 incremental 模式；checkpoint interval 是多少，tps
大概多少。这些数据用于评估 rocksdb incremental 下 checkpoint size 的大小
2）看一下 hdfs 上的 checkpoint 路径占用大小是否有变化
3）像 hk__lrzy 说的那样，state 是否没有清理（这里还需要看下 window 相关的）
Best,
Congxian
hk__lrzy 于2020年9月11日周五
Hi
从错误栈看是 Wrong FS:
file:///data/emr_flink_savepoint_share/savepoint-0705a3-09a2f171a080/e2f63448-eed9-4038-a64a-e874a1a99ba1,
expected: hdfs://flink-hdfs 这个导致的，你能把 savepoint 写到 hdfs://flink-hdfs 这个集群吗？
Best,
Congxian
hk__lrzy 于2020年9月11日周五 下午2:46写道：
> 代码是不是主动设置过stagebackend的地址呢
>
>
>
> --
Hi
从 retain checkpoint 恢复可以参考文档[1]
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/ops/state/checkpoints.html#%E4%BB%8E%E4%BF%9D%E7%95%99%E7%9A%84-checkpoint-%E4%B8%AD%E6%81%A2%E5%A4%8D%E7%8A%B6%E6%80%81
Best,
Congxian
sun <1392427...@qq.com> 于2020年9月3日周四 下午4:14写道：
>
从之前的 checkpoint/savepoint 恢复的话，加上 -n 或者 --allowNonRestoredState
是可以恢复的，不过需要注意如何保证从 *特定* 的 offset 进行恢复
Best,
Congxian
Paul Lam 于2020年9月3日周四 上午11:59写道：
> 可以，保证 RokcetMQ source 算子的 uid 和原本的 Kafka source 算子的 uid 不同就行。
> 另外启动要设置参数 -n 或 —allowNonRestoredState 。
>
> Best,
> Paul Lam
>
> > 2020年9月2日
Hi
按理说 checkpoint/savepoint 有的数据，正常恢复后是可以读取到的。
1 正常从 checkpoint/savepoint 恢复了吗？
2 获取 state 的时候，key 是同一个 key 吗？
Best,
Congxian
Liu Rising 于2020年9月3日周四 上午9:28写道：
> 版本: 1.9
>
> 问题：
> 当从savepoint或者checkpoint恢复flink job时，发现部分keyedState中的数据丢失。
> 这里我们使用的是ListState，里面存储的是ObjectNode(Jackson
Hi
从火焰图看，RocksDB#get 操作占用的时间较多，contains 会调用 RocksDB 的 get 函数
1. 你使用的是哪个版本的 Flink？
2. 不同 subtask 之间的数据是否均匀呢？这里主要想知道调用 RocksDB 的 get 函数调用频次是否符合预期
3. 如果我理解没错的话，有 snappy 的压缩，这个会有 IO 的操作（也就是从磁盘 load 数据），可能还需要看下为什么这个
subtask 的数据大量落盘
Best,
Congxian
fanrui <836961...@qq.com>
Hi
可以看下 本地的 log 文件($FLINK_DIR/log 下）以及 jm log，看看有没有什么异常情况
Best,
Congxian
462329521 <462329...@qq.com> 于2020年9月1日周二 下午3:41写道：
>
> Hello,我在flink1.8使用flinkcancel命令后，任务处于canceled状态，但是yarn上仍在running，请问是什么情况
Hi
如果我理解没错的话，这种 单 key 热点的问题，需要算 中位数（无法像 sum/count
这样分步计算的），只能通过现在你写的这种方法，先分布聚合，然后最终再计算中位数。不过或许可以找找数学方法，看有没有近似的算法
Best,
Congxian
赵一旦 于2020年9月1日周二 上午10:15写道：
> （1）url理论上足够多，也足够随机。而并行度比如是30，url理论上是万、十万、百万、千万级别，理论上不会出现数据倾斜吧。
> （2）如果的确有倾斜，那么你那个方法我看不出有啥用，我看你好像是全缓存下来？这没啥用吧。
>
Hi
checkpoint 只能保证 state 的 exactly once，但是单条数据可能重复处理多次，如果是 sink
输出多次的话，或许你可以看一下 TwoPhaseCommitSinkFunction 相关的，这篇文章有一个相关的描述[1]
[1]
https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
Best,
Congxian
Kevin Dai <154434...@qq.com> 于2020年8月28日周五 上午9:44写道：
Hi
Checkpoint 的 size 取决于 state 的大小（如果是 RocksDBStateBackend + 增量
checkpoint，界面看到的 Checkpointed Data Size 是增量大小[1])。如果你把 BloomFilter 存到 State
中的话，那么需要看看最终在 State 中存储的内容大小。
如果有疑问的话，可以分析一下分别使用 bloomfilter 和 mapstate，在相同的数据更新后，state 中的数据分别有多少
[1]
下你的代码（可以把业务逻辑进行简化），或许大家可以帮你看看
> Best,
> Congxian
>
>
> Robert.Zhang <173603...@qq.com 于2020年8月26日周三 上午11:43写道：
>
> Hi Congxian,
>
> 开了更多日志观察了下，是由于iteration source的barrier无法接收到导致的。
> 该barrier无法接收到，导致下游也无法拿到由该operator传递的barrier。因此checkpoint一直无法成功。
> 比较奇
Hi
从异常看，可能是类冲突了，或许是有两个版本的 `org.apache.orc.TypeDescription` 依赖，可以排除或者 shade
一下相关的 class
Best,
Congxian
liya...@huimin100.cn 于2020年8月27日周四 下午8:18写道：
> Exception in thread "main" java.lang.NoSuchMethodError:
>
Congratulations Dian
Best,
Congxian
Xintong Song 于2020年8月27日周四 下午7:50写道：
> Congratulations Dian~!
>
> Thank you~
>
> Xintong Song
>
>
>
> On Thu, Aug 27, 2020 at 7:42 PM Jark Wu wrote:
>
> > Congratulations Dian!
> >
> > Best,
> > Jark
> >
> > On Thu, 27 Aug 2020 at 19:37, Leonard Xu wrote:
Hi
对于开启 Checkpoint
之后导致反压的情况，如果希望在现在的基础上进行优化的话，则需要找到反压的原因是什么，可以尝试从最后一个被反压的算子开始排查，到底什么原因导致的，排查过程中，或许
Arthas[1] 可以有一些帮助
另外比较好奇的是，为什么反压会导致你的作业挂掉呢？作业挂掉的原因是啥呢
[1] https://github.com/alibaba/arthas
Best,
Congxian
Yun Tang 于2020年8月26日周三 上午11:25写道：
> Hi
>
> 对于已经改为at least
rrier。因此checkpoint一直无法成功。
> 比较奇怪的一点是，这是在我停止数据源发送的情况下出现的。我发送一条经过这个operator的数据，
> 那么从日志中看，立刻获得了barrier，checkpoint可以迅速完成。
>
>
>
> 为何会出现这个情况呢？什么情况下流里面没有数据会导致无法checkpoint成功呢？
>
>
>
>
>
> ---原始邮件---
> *发件人:* "Congxian Qiu"
> *发送时间:* 2020年8月25日(周二) 下午
Thanks ZhuZhu for managing this release and everyone else who contributed
to this release!
Best,
Congxian
Xingbo Huang 于2020年8月26日周三 下午1:53写道：
> Thanks Zhu for the great work and everyone who contributed to this release!
>
> Best,
> Xingbo
>
> Guowei Ma 于2020年8月26日周三 下午12:43写道：
>
>> Hi,
>>
Hi
据我所知，在作业启动之后，是无法改变数据的分法规则的，也就是说没办法做到这个要求。
Best,
Congxian
Sun_yijia 于2020年8月26日周三 下午2:17写道：
> 在做反压相关的代码，想请教各位大佬。
>
>
> 有一个分支节点，分支后面有两个节点A和B。假设A节点出现了反压，B节点负载空闲。
> 我想让B节点帮A节点做一些计算，这样B节点就能够缓解一部分A节点的压力。
>
>
> 有什么方法能让Flink在运行过程中，把接下来要发给A节点的数据发送给B节点吗？
完成，web界面上 iteration source的checkpoint始终无法完成。
> 官方文档对于在iterative
> stream的checkpoint没有更详细的说明。对于loop中的数据丢失可以理解。但是checkpoint无法成功不是特别能理解。
> 按照我对于chandylamport算法的理解，上游operator的barrier应该是直接给到了下游
> ，不应该存在无法拿到barrier的情况才对。不知道这是什么原因导致的
>
> ---原始邮件---
> 发件人: "Congxian Qiu" 发送时
8月19日周三 下午3:03写道：
>
> 感谢邱老师，这个我查看了一下没有这个文件的，跟现在运行的相同任务的正常执行的chk目录下的文件相比这个chk-167目录下的文件数少了很多，我们当时是看着cp执行完成之后cancel了任务然后
> 从hdfs上查到这个目录cp路径去重启的任务
>
> Congxian Qiu 于2020年8月19日周三 下午2:39写道：
>
> > Hi
> >1 图挂了
> > 2 你到 hdfs 上能找到
Hi
理论上第一次能启动，后续的 failover 也应该是可以正常恢复的。你这边是稳定复现吗？如果能够稳定复现的话，有可能是 bug
Best,
Congxian
xiao cai 于2020年8月20日周四 下午2:27写道：
> Hi：
> 感谢答复，确实是个思路。
>
> 不过个人感觉能够在启动第一个container的时候，将本地的lib中的jar文件上传到hdfs中，然后后续的failover的container能够统一从hdfs中获取，这样应该就不会有这个问题了。貌似社区在1.11版本针对jar的拷贝做了优化，我还在看这方面的内容。有进展再交流。
>
>
>
Hi
分享一篇讲解增量 checkpoint 的文章[1]
[1]
https://ververica.cn/developers/manage-large-state-incremental-checkpoint/
Best,
Congxian
Yun Tang 于2020年8月21日周五 上午12:09写道：
> Hi
>
> 增量checkpoint与web界面的信息其实没有直接联系，增量checkpoint的信息记录由CheckpointCoordinator中的SharedStateRegistry[1]
> 进行计数管理，而保留多少checkpoint则由
Hi
从报错看 CountDownLatch 这个方法无法 serializable，这个 class 没有实现 Serializable
接口。你可以按照这里的方法[1] 尝试解决下
[1]
https://stackoverflow.com/questions/4551926/java-io-notserializableexception-while-writing-serializable-object-to-external-s/4552014
Best,
Congxian
引领 于2020年8月24日周一 下午3:34写道：
>
> 使用场景：FLink
Hi
比较好奇你为什么在 Blink 分支做测试，而不是用最新的 1.11 做测试呢？
Best,
Congxian
柯四海 <2693711...@qq.com> 于2020年8月24日周一 下午5:58写道：
> Hi 大家好，
> 我用github上Blink分支(1.5)编译的flink来运行一些实时任务，发现Taskmanager
> 因为内存超了container限制被yarn kill.
> 有没有人有比较好的问题定位方案？
>
> 尝试过但是还没有解决问题的方法：
> 1. 尝试增加taskmanager内存
> 修改： 从8G 提高到
Hi
从报错 ”Exceeded checkpoint tolerable failure threshold“ 看，你的 checkpoint
有持续报错，导致了作业失败，你需要找一下为什么 checkpoint 失败，或许这篇文章[1] 可以有一些帮助
另外从配置看，你开启了 unalign checkpoint，这个是上述文章中暂时没有设计的地方。
[1] https://zhuanlan.zhihu.com/p/87131964
Best,
Congxian
Robert.Zhang <173603...@qq.com> 于2020年8月21日周五
Hi
StateBackend 可以理解为 一个 KV 存储加上一个 snapshot 过程，其中 snapshot 过程负责将当前 KV
存储的数据进行备份。理论上任何的 KV 存储都是有可能作为 StateBackend 的，不过增加一种 StateBackend 的话，需要实现相应的
snapshot/restore 逻辑。
但是在多个 Flink 作业中实现共享的 state 这个在 Flink 中是不支持的。
Best,
Congxian
wxpcc 于2020年8月21日周五 下午6:33写道：
> 项目里有部分需要进行状态共享的需求，多个flink
MasterService(DefaultJobMasterServiceFactory.java:98)
> at
> org.apache.flink.runtime.jobmaster.factories.DefaultJobMasterServiceFactory.createJobMasterService(DefaultJobMasterServiceFactory.java:40)
> at
> org.apache.flink.runtime.jobmaster.JobManagerRunner.(JobManagerRunner.
Hi
从栈看应该是 deserialize 的时候出错了，另外 kryo 可以，Pojo 不行，能否检查下，是否满足 POJO 的一些要求[1]呢?
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/state/schema_evolution.html#pojo-types
Best,
Congxian
shizk233 于2020年8月18日周二 下午4:09写道：
> Hi all,
>
>
Hi
1.9 上是否加 -d 应该会使用不同的模式来启动作业 (perjob 还是
session），这两个模式下的行为应该是不完全一致的，具体的可以看下这里[1]
[1]
https://github.com/apache/flink/blob/5125b1123dfcfff73b5070401dfccb162959080c/flink-clients/src/main/java/org/apache/flink/client/cli/CliFrontend.java#L211
Best,
Congxian
bradyMk 于2020年8月19日周三
Hi
如果你的算子有改变的话，想从 checkpoint/savepoint 恢复，需要添加
`--allowNonRestoredState`，这样可以忽略掉那些不在新
job 中的算子（就算逻辑一样，uid 不一样也会被忽略掉的），具体的可以参考文档[1]
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/state/savepoints.html#what-happens-if-i-delete-an-operator-that-has-state-from-my-job
Best,
Hi
你的 Flink 是哪个版本，期望的行为是什么样的？
从你给的日志看，是没有 这个 class，这个 是在你放到 lib 下的某个 jar
包里面吗？另外你这个作业第一次运行的时候失败，还是中间中间 failover 之后恢复回来的时候失败呢？
Best,
Congxian
xiao cai 于2020年8月19日周三 下午12:50写道：
> 如题：link on Yarn 启动任务，任务重启后貌似taskmanager中会丢失lib中的jar文件
>
>
> 我的任务时kafka source -> hbase sink
>
>
>
Hi
像我之前说的那样，加 -d 和不加 -d 使用的是不同的模式启动作业的。从你的报错栈来看，应该是类冲突了。你可以看下这个文档[1] 看看能否帮助你
java.lang.NoSuchMethodError:
art 会开启一个事务 编号proccess 用这个事务 编号
> 去做预处理(赞一批处理一次，并把这一次处理结果下发，给下游做事务提交)， submit 收到上游批处理的结果 用 同样的事务编号去提交
>
>
> Congxian Qiu 于2020年8月17日周一 上午10:42写道：
>
> > Hi
> > 上游 snapshot 的逻辑和下游收到之前的 notifyCheckpointComplete
> > 之间是没有必然联系的，所以这个从理论上是不保证先后顺序的。
上午11:22写道：
> 在我们自研的开发平台上提交任务用的detach模式，提交完之后就看不到其他日志了，这个问题当天出现了两次，是不是使用增量cp会存在这个恢复失败的情况
>
> Congxian Qiu 于2020年8月17日周一 上午10:39写道：
>
> > Hi
> >你还有失败作业的 JM 和 TM
> > 日志吗？如果有的话可以看一下这两个日志来确定为什么没有恢复成功。因为你说代码未作任何改变，然后恢复失败，这个还是比较奇怪的。
> > Best,
> > Congxian
&g
Hi
如果我理解没错的话，是否添加 -d 会使用不同的模式启动作业（PerJob 和 Session
模式），从错误栈来看猜测是版本冲突了导致的，你有尝试过最新的 1.11 是否还有这个问题吗？
Best,
Congxian
bradyMk 于2020年8月14日周五 下午6:52写道：
> 请问大家：
> 我采用如下命令提交：
> flink run \
> -m yarn-cluster \
> -yn 3 \
> -ys 3 \
> -yjm 2048m \
> -ytm 2048m \
> -ynm flink_test \
> -d \
> -c
这个问题和下面这个问题[1] 重复了，在另外的邮件列表中已经有相关讨论
[1]
http://apache-flink.147419.n8.nabble.com/Flink-FINISHED-Checkpoint-td6008.html
Best,
Congxian
yulu yang 于2020年8月14日周五 下午1:05写道：
> 对了，我这个flink作业和和分组都是新创建，不存在抽取历史。
>
> 杨豫鲁 于2020年8月13日周四 下午3:33写道：
>
> > 请教大家一个我最近在配置Flink流的过程中遇到问题，
> >
> >
>
Hi 吴磊
请问你们有比较过使用 Redis 和 broadcast state 在你们场景下的区别吗？是什么原因让你们选择 Redis 而不是
BroadcastState 呢？
Best,
Congxian
吴磊 于2020年8月14日周五 下午3:39写道：
> 在我们的生产环境最常用的做法都是通过维表关联的方式进行赋值的；
> 或者可以先将字典数据写进redis，然后再在第一次使用的时候去访问redis，并加载到State中。
>
>
>
>
> --原始邮件--
> 发件人:
>
Hi
上游 snapshot 的逻辑和下游收到之前的 notifyCheckpointComplete
之间是没有必然联系的，所以这个从理论上是不保证先后顺序的。
Best,
Congxian
key lou 于2020年8月16日周日 下午9:27写道：
> 各位大佬：
>在如下代码中： FCombine 执行snapshot collect 发送数据之后如果不执行sleep 则 FSubmit
> 在执行 notifyCheckpointComplete 方法时，list 集合 ls 为空。
> 如果在 FCombine 执行snapshot
Hi
你还有失败作业的 JM 和 TM
日志吗？如果有的话可以看一下这两个日志来确定为什么没有恢复成功。因为你说代码未作任何改变，然后恢复失败，这个还是比较奇怪的。
Best,
Congxian
Yang Peng 于2020年8月17日周一 上午10:25写道：
> 好的 感谢
>
> JasonLee <17610775...@163.com> 于2020年8月14日周五 下午9:22写道：
>
> > hi
> >
> > 没有日志不太好定位失败的原因 但是没有设置uid的话 是有可能重启失败的 建议还是都设置uid最好
> >
> >
> >
> > --
> >
Hi
这应该是个已知问题[1] 在 1.11.2 和 1.12 中已经修复
[1] https://issues.apache.org/jira/browse/FLINK-18710
Best,
Congxian
郭华威 于2020年8月13日周四 上午11:05写道：
> 你好，请教下：
> flink1.11.1 flink on yarn 任务启动报错：
>
>
> 启动命令：
> /opt/flink-1.11.1/bin/flink run -p 4 -ys 2 -m yarn-cluster -c
> yueworld.PVUV.MyPvUv
Hi
不好意思，上一份邮件没有说完就发送出去了。
如果你希望把从其他地方读入 字典表，然后在 flink 中使用，或许可以看看 broadcast state[1]
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/state/broadcast_state.html
Best,
Congxian
Congxian Qiu 于2020年8月13日周四 下午2:00写道：
> Hi
> 现在 checkpoint/savepoint 需要所
Hi
现在 checkpoint/savepoint 需要所有算子都处于 RUNNING 状态，不过社区也有一些 issue
希望能够优化这个问题[1][2]
[1] https://issues.apache.org/jira/browse/FLINK-2491
[2] https://issues.apache.org/jira/browse/FLINK-18263
Best,
Congxian
yulu yang 于2020年8月13日周四 下午1:49写道：
> 请教大佬一个我最近在配置Flink流的过程中遇到问题，
>
>
Hi
你能拿到 memory 的 dump 吗？OOM 可能需要看一下 memory 的 dump 才能更好的确定是什么问题
Best,
Congxian
ReignsDYL <1945627...@qq.com> 于2020年8月11日周二 下午4:01写道：
> 各位好，
> Flink 1.10，集群在运行过程中，堆外内存一直在不断增加，内存就被慢慢耗尽，导致任务挂掉，请问是什么原因啊？
>
>
>
> --
> Sent from: http://apache-flink.147419.n8.nabble.com/
hi
1 checkpoint/savepoint 可以理解为将 状态备份到远程存储，恢复的时候会通过 operator 的 uid 来恢复
state，如果你确定不希望某些 operator 的 state 不进行恢复的话，或者使用不同的 uid
可以达到你的需求，具体的可以看一下这个文档的内容[1]
2 合并的时候如果想把 savepoint/checkpoint 用起来，还是需要修改 checkpoint/savepoint
的内容，或者你可以试试 state processor api[2]
[1]
amp;gt;
> gt; amp;gt; 刚启动的时候，如下：
> gt; amp;gt; <
>
> http://apache-flink.147419.n8.nabble.com/file/t793/6.pngamp;amp;gt;
> gt; amp;gt;
> gt; amp;gt; 18分钟后，如下：
> gt; amp;gt; <
>
> http://apache-flink.147419.n8.nabble.com/file/t793/9.pngamp;amp;gt;
> gt;
Hi xuhaiLong
请问你这个作业在这个版本是是必然出现 NPE 问题吗？另外 1.10 之前的版本有出现过这个问题吗？
Best,
Congxian
xuhaiLong 于2020年8月7日周五 下午3:14写道：
> 感谢回复！我这边的确是这个bug 引起的
>
>
> On 8/7/2020 13:43，chenkaibit wrote：
> hi xuhaiLong,看日志发生的 checkpoint nullpointer 是个已知的问题，具体可以查看下面两个jira。
> 你用的jdk版本是多少呢？目前发现使用 jdk8_40/jdk8_60 +
Hi
对于 event time 的处理来说，不建议注册 timer 的时候使用 System.currentTimeMillis()
这种系统时间，这两个时间可能会不一样，可以使用 TimerService 中的 currentWatermark 表示当前的 event time
Best,
Congxian
jsqf 于2020年8月6日周四 下午9:53写道：
> 试试 重写 onTimer 方法
> 可以参考
>
>
> > gt;
> > gt;
> > gt;
> > gt; hi，您好：
> > gt; 我改回增量模式重新收集了一些数据：
> > gt; 1、数据处理速度：3000条每秒，是测试环境的，压力比较稳定，几乎没有波动
> > gt; 2、checkpoint是interval设置的是5秒
> > gt; 3、目前这个作业是每分钟一个窗口
> > gt; 4、并行度设置的1，使用on-yarn模式
> > gt;
> > g
钟一个窗口
> gt; 4、并行度设置的1，使用on-yarn模式
> gt;
> gt; 刚启动的时候，如下：
> gt; <
> http://apache-flink.147419.n8.nabble.com/file/t793/6.pngamp;gt;
> gt;
> gt; 18分钟后，如下：
> gt; <
> http://apache-flink.147419.n8.nabble.com/file/t793/9.pngamp;gt;
> gt;
> gt; checkpoints设置：
al设置的是5秒
> 3、目前这个作业是每分钟一个窗口
> 4、并行度设置的1，使用on-yarn模式
>
> 刚启动的时候，如下：
> <http://apache-flink.147419.n8.nabble.com/file/t793/6.pnggt;
>
> 18分钟后，如下：
> <http://apache-flink.147419.n8.nabble.com/file/t793/9.pnggt;
>
> checkpoints设置：
> <http://apache-f
Hi 张洋
如果我理解没错的话，现在 Flink
无法严格保证消费了一条数据，等待上一条数据处理完成，然后再消费下一条数据的。如果想做到这个需求，需要用户做一些事情。
你说的第 2 点中是否处理完成，这个能否依赖第三方服务，在 sink（或者最后一个 operator） 处理完成之后做通知呢？
checkpoint 如果正常完成的话，那就是没有异常的，不过 checkpoint 没法完全保证一条数据一条数据的处理
Best,
Congxian
Bruce 于2020年8月5日周三 上午9:33写道：
>
Hi
或许你可以看一下 Flink 作业的 JM 是不是还在运行着？
Best,
Congxian
bradyMk 于2020年8月4日周二 上午11:38写道：
> 请教大家：
>
> flink1.9.1任务已经fail掉了，但在yarn上这个application还是在running，且yarn上分配的资源变成了1，程序中用的是固定延迟重启策略，请问有人知道任务挂掉但yarn上一直在running是什么原因么？
> <
>
共有 244 项搜索結果，以下是第 1 - 100 matches
Mail list logo
