[ANNOUNCE] Apache Flink 1.17.2 released

The Apache Flink community is very happy to announce the release of Apache Flink 1.17.2, which is the second bugfix release for the Apache Flink 1.17 series. Apache Flink® Is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams. Flink

Re: [ANNOUNCE] Flink Table Store Joins Apache Incubator as Apache Paimon(incubating)

Congratulations! Unlike other data-lakes, Paimon might be the first one to act as a stream-first (not batch-first) data-lake. Best Yun Tang From: Xianxun Ye Sent: Tuesday, March 28, 2023 10:52 To: d...@flink.apache.org Cc: Yu Li ; user ; user-zh ; d

Re: [ANNOUNCE] FRocksDB 6.20.3-ververica-2.0 released

Thanks Yuanfei for driving the frocksdb release! Best Yun Tang From: Yuan Mei Sent: Tuesday, January 31, 2023 15:09 To: Jing Ge Cc: Yanfei Lei ; d...@flink.apache.org ; user ; user-zh@flink.apache.org Subject: Re: [ANNOUNCE] FRocksDB 6.20.3-ververica-2.0

Re: 怀疑源码中的一个方法是never reached code

Hi，育锋 我觉得你的分析应该是没问题的。可以创建一个ticket来修复该问题。另外，关于代码实现的具体讨论，建议在dev邮件列表讨论。 祝好 唐云 From: 朱育锋 Sent: Tuesday, June 14, 2022 19:33 To: user-zh@flink.apache.org Subject: 怀疑源码中的一个方法是never reached code Hello Everyone

Re: 1.13.5版本sql大小64k限制bug

Hi 请使用英文在dev社区发送邮件。另外关于使用方面的问题，建议向user-zh 频道发送，已经帮你转发到相关邮件列表了。 祝好 唐云 From: Lose control .／ <286296...@qq.com.INVALID> Sent: Tuesday, May 24, 2022 9:15 To: dev Subject: 1.13.5版本sql大小64k限制bug 请问各位大神，1.13.5版本sql大小64k限制如何修改啊？谢谢

Re: RocksDB 读 cpu 100% 如何调优

Hi, RocksDB 的CPU栈能卡在100%，很有可能是大量解压缩 index/filter block导致的，可以enable partition index/filter [1] 看看问题是否解决。 相关内容也可以参考我之前线下做过的分享 [2] [1] https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#state-backend-rocksdb-memory-partitioned-index-filters [2]

Re: Re:flink s3 checkpoint 一直IN_PROGRESS(100%)直到失败

Hi 一般是卡在最后一步从JM写checkpoint meta上面了，建议使用jstack等工具检查一下JM的cpu栈，看问题出在哪里。 祝好 唐云 From: Sun.Zhu <17626017...@163.com> Sent: Tuesday, March 8, 2022 14:12 To: user-zh@flink.apache.org Subject: Re:flink s3 checkpoint 一直IN_PROGRESS(100%)直到失败 图挂了

Re: 状态初始化

Hi, 这个需求在社区里面称之为 state bootstrapping， 以前在state processor API没有引入时，还有第三方的工具 bravo [1]。 我理解你的需求完全可以有state processor API完成，生成一个savepoint，由新作业消费。目前社区也在考虑支持生成native savepoint，用以加快生成速度 [2] [1] https://github.com/king/bravo [2] https://issues.apache.org/jira/browse/FLINK-25528 Best Yun Tang

Re: Re: flink sql支持细粒度的状态配置

Hi, 如果你们可以自己实现一套SQL语句到jobgraph的预编译转换IDE，然后在IDE中可以手动配置jobgraph每个算子的配置，应该是可以达到你们的目的 （可能还需要结合细粒度调度模式）。 祝好 唐云 From: gygz...@163.com Sent: Thursday, December 9, 2021 16:14 To: user-zh Subject: 回复: Re: flink sql支持细粒度的状态配置 Hi Yun Tang 感谢你的回复，我们在调研的过程中也发现，正如你所说的生成的

Re: flink sql支持细粒度的状态配置

Hi 你好， 我认为这是一个很好的需求，对于data stream以及python API来说，state TTL都是通过API逐个配置的，你的需求就可以直接满足。但是对于SQL来说，由于相同的SQL语句，不同优化器其生成的执行plan可能会差异很大，很难对某个operator内的state进行TTL进行配置，可能一种方式是增加一些SQL的优化hint，对于你示例中的join语句和groupBy 的count语句配以不同的TTL，但是目前Flink SQL尚未支持该功能。 祝好 唐云 From:

Re: 检查点和保存点

Hi checkpoint 以及 savepoint是否可以生效取决于相关source的实现，Kafka这种是支持replay非常好的source，至于file reader，目前 split file reader [1] 相关的实现是支持 容错的 [1] https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/sources/#the-split-reader-api 祝好 唐云 From: lei-tian

Re: MongoDB sink

Hi, 具体问题建议直接在相关ticket上进行讨论，邮件列表上可能相关人士没有注意到。 祝好 唐云 From: 不许人间见白头 Sent: Wednesday, November 10, 2021 22:28 To: user-zh Subject: MongoDB sink 你好， 请问一下，关于New feature: FLINK-24477 预计什么时候创建PR呢？

Re: flink 以阿里云 oss 作为 checkpoint cpu 过高

Hi 可以使用jstack，async profiler [1] 等工具勘察一下checkpoint期间的CPU栈。oss需要先写本地再上传，确实可能CPU消耗多一些，但是明显高很多有一些超出预期。 [1] https://github.com/jvm-profiling-tools/async-profiler 祝好 唐云 From: Lei Wang Sent: Tuesday, October 19, 2021 14:01 To: user-zh@flink.apache.org Subject: Re:

Re: 关于作业失败从checkpoint重启，触发了过期的窗口计算

Hi， 先问个版本问题，你的Flink版本是1.3 而不是1.13？ Checkpoint里面会存储timer，所以重启之后会触发窗口的计算，但确实这种一天的窗口累计有点太多了，除非你的作业存在比较严重的反压，导致checkpoint内积攒了大量没有触发的timer。 祝好 唐云 From: claylin <1012539...@qq.com.INVALID> Sent: Friday, October 29, 2021 11:33 To: user-zh Subject:

Re: Re:re:Re: 回复：Flink SQL官方何时能支持redis和mongodb连接器？

t; 写道： >我司基于最新提供流批一体的接口，实现了mongodb的连接器，支持source和sink，实现了控制批量插入频率、控制缓存批的数据量和mongo文档对象和java对象转换，同时还可选择批量更新，并且使用的mongo最新异步驱动，后期我还会不断优化性能，看大佬能否推动一下，把这个连接器贡献给社区 > > >-- 原始邮件 ------ >发件人: "Yun Tang"; >发件时间: 2021-09-22 10:55 >收件人: "user-z

Re: 回复：Flink SQL官方何时能支持redis和mongodb连接器？

Hi, 其实目前Flink社区并不是那么欢迎新增官方支持的connector，主要原因就是社区的开发人员有限，没有精力维护太多的connector，尤其是一些connector的实现需要一定的相关背景，但很难保证review代码的开发人员具有相关背景，毕竟大家都需要为自己approve的代码负责。 你可以在 flink-packages [1] 里面找一下，或者考虑自己实现并维护（基础实现应该是复杂度不高的）。 [1] https://flink-packages.org/ 祝好 唐云 From: 黑色

Re: flink oss ha

ir=oss://bucket-logcenter/flink-state/flink-session-recovery \ -Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-oss-fs-hadoop-1.13.2.jar \ -Dcontainerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS=flink-oss-fs-hadoop-1.13.2.jar -----邮件原件- 发件人: Yun Tang 发送时间: 2021年8月30日 11:36 收件人:

Re: flink oss ha

Hi, 你好，图片无法加载，可以直接粘贴文字出来 祝好 唐云 From: dker eandei Sent: Friday, August 27, 2021 14:58 To: user-zh@flink.apache.org Subject: flink oss ha 您好: 看文档OSS可以用作 FsStatebackend，那么Flink on k8s 做高可用时，high-availability.storageDir可以配置成oss吗，我试了下，报以下错误:

Re: table.exec.state.ttl

Hi 航飞 可以参照[1] 看是不是类似的问题 [1] https://issues.apache.org/jira/browse/FLINK-23721 祝好 唐云 From: 李航飞 Sent: Thursday, August 26, 2021 19:02 To: user-zh Subject: table.exec.state.ttl Configuration conf = new Configuration();

[ANNOUNCE] Apache Flink 1.13.2 released

are available in Jira: https://issues.apache.org/jira/secure/ReleaseNote.jspa?version=12350218==12315522 We would like to thank all contributors of the Apache Flink community who made this release possible! Regards, Yun Tang

Re: 1.14啥时候出呀

Hi Flink-1.13.2 的jar包正在同步到给个maven仓库，顺利的话，明天就可以正式announce了。 祝好 唐云 From: Jingsong Li Sent: Wednesday, August 4, 2021 16:56 To: user-zh Subject: Re: 1.14啥时候出呀 1.14还有1-2个月 1.13.2马上就出了，估计明天或后天或周一 On Wed, Aug 4, 2021 at 4:48 PM yidan zhao wrote: >

Re: flink大窗口性能问题

目前Flink社区版RocksDB尚不支持ARM架构机器。使用RocksDB的话，内存均是堆外管理，与JVM的堆上内存无关。 另外，有个题外话，你们是云上产品还是自建了ARM集群？有点好奇目前国内的ARM集群使用率情况。 祝好 唐云 From: Wanghui (HiCampus) Sent: Thursday, July 15, 2021 11:33 To: user-zh@flink.apache.org Subject: Re: flink大窗口性能问题 我在aarch64 + jre

Re: flink 触发保存点失败

Hi, 这个看上去是client触发savepoint失败，而不是savepoint本身end-to-end执行超时。建议对照一下JobManager的日志，观察在触发的时刻，JM日志里是否有触发savepoint的相关日志，也可以在flink web UI上观察相应的savepoint是否出现在checkpoint tab的历史里面。 祝好 唐云 From: 仙剑……情动人间 <1510603...@qq.com.INVALID> Sent: Tuesday, July 13, 2021 17:31 To:

Re: local运行模式下不会生成checkpoint吗？

Hi 只要enable了checkpoint，一定会生成checkpoint的，这与你的运行模式无关。可以检查一下日志，看看JM端是否正常触发了checkpoint 祝好 唐云 From: casel.chen Sent: Tuesday, June 29, 2021 9:55 To: user-zh@flink.apache.org Subject: local运行模式下不会生成checkpoint吗？ 我在本地使用local运行模式运行flink sql，将数据从kafka写到mongodb，mongodb

Re: Flink 1.10 内存问题

Hi, 有可能的，如果下游发生了死锁，无法消费任何数据的话，整个作业就假死了。要定位root cause还是需要查一下下游的task究竟怎么了 祝好 唐云 From: Ada Luna Sent: Tuesday, July 6, 2021 12:04 To: user-zh@flink.apache.org Subject: Re: Flink 1.10 内存问题 反压会导致整个Flink任务假死吗？一条Kafka数据都不消费了。持续几天，不重启不恢复的 Yun Tang 于2021年7月6日周二 上午11

Re: Flink 1.10 内存问题

Hi, LocalBufferPool.requestMemorySegment 这个方法并不是在申请内存，而是因为作业存在反压，因为下游没有及时消费，相关buffer被占用，所以上游会卡在requestMemorySegment上面。 想要解决还是查一下为什么下游会反压。 祝好 唐云 From: Ada Luna Sent: Tuesday, July 6, 2021 10:43 To: user-zh@flink.apache.org Subject: Re: Flink 1.10 内存问题

Re: 中文教程更新不及时问题

Hi Kevin, 欢迎来到Apache Flink开源社区！ 因为开源社区的工作，一些参与者很多时候都是工作时间之外参与的，可能难免遇到进度更新不及时，或者长时间不再活跃的问题。 非常欢迎您在相关JIRA ticket下面评论和申请权限创建PR，社区一直都欢迎每一位贡献者，对于文档的维护尤其是中文文档的翻译也是非常需要的，如果有任何想要贡献的部分，欢迎直接去JIRA ticket下面、github PR下面评论，或者直接创建相关ticket。 祝好 唐云 From: pang fan Sent:

Re: rocksdb对比filestatebackend

Hi Yidan, 1. 是否我从FileStateBackend切换到RocksDB其实性能也不会降低很多呢？ 2. 这个涉及到RocksDB这种LSM架构的DB读写路径了，即使逻辑数据量可以全部存储在内存中，由于RocksDB的write buffer默认会存储相同key的不同value，而且checkpoint时候仍然会触发flush，很难避免数据落盘，数据落盘之后的读路径肯定没有Flink的内存state backend性能好，二者性能还是有些差异的，不过实际生产中可能不需要 FsStateBackend 那么高的性能。 1.

Re: Flink state evolution with avro

Hi, 你可以参照社区的 state-evolution的 E2E 测试代码 [1], 整个程序就是使用的avro作为相关类的声明工具。 [1] https://github.com/apache/flink/tree/master/flink-end-to-end-tests/flink-state-evolution-test/src/main 祝好 唐云 From: casel.chen Sent: Friday, June 11, 2021 8:13 To:

Re: 关于flink sql的kafka source的开始消费offset相关问题。

hi, 本质上来说，你的做法有点hack其实不推荐，如果非要这么做的话，你还可以通过 numRestarts [1] 的指标来看重启了多少次。 [1] https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/ops/metrics/#availability 祝好 唐云 From: yidan zhao Sent: Friday, June 4, 2021 11:52 To: user-zh Subject: Re: 关于flink

Re: Flink Sql 的/checkpoint/shared/文件夹大小不断增长，源数据没有数据激增，应该如何控制？

Hi, 没有被引用的文件可能也不是完全无用的，可能是当前pending checkpoint正在上传的，所以还需要比较一下那些不在checkpoint meta内的文件的修改时间戳，可能比你分析的complete checkpoint的时间戳要大。 总体上来说，我不认为这个问题是一个bug，这个是LSM架构的DB的空间放大问题。如果你对空间放大非常担心的话，可以启用 dynamic level [1] 来严格控制空间放大，不过这个可能会影响写放大和读放大，导致性能受到一定影响。 [1]

Re: Flink Sql 的/checkpoint/shared/文件夹大小不断增长，源数据没有数据激增，应该如何控制？

Hi， 增量checkpoint上传的是sst文件本身，里面可能有一部分空间是被无用数据占据的，你可以理解成增量checkpoint上传的是受到空间放大影响的RocksDB的数据，如果因为单机的数据量较小，没有及时触发compaction的话，确实存在整个远程checkpoint目录数据大于当前实际空间的情况。而关闭增量checkpoint，上传的其实是与savepoint格式一样的kv数据对，Flink会遍历整个DB，将目前有效的数据写出到远程。所以你关闭增量checkpoint，而发现checkpoint目录保持恒定大小的话，说明真实有效数据的空间是稳定的。

Re: Flink Sql 的/checkpoint/shared/文件夹大小不断增长，源数据没有数据激增，应该如何控制？

Hi, 先确定一下，你的 idleStateRetention 是 3600秒？其次，要想看是否所有数据均有用，可以利用 Checkpoints.loadCheckpointMeta [1] 去加载你所保留的checkpoint目录下的 _metadata 文件，然后与当前checkpoint目录下的文件作对比，看是否存在大量的未删除旧文件。 目前仅凭你的描述和一段SQL代码其实很难判断。 可能存在的原因有： 1. 单次checkpoint文件数目过多，JM单点删除跟不上相关速度 2. 整体checkpoint

Re: Flink upgraded to version 1.12.0 and started from SavePoint to report an error

Hi BaseRowSerializer 已经在Flink-1.11 时候改名成 RowDataSerializer了，即使用 state-processor-API 也没办法处理当前不存在的类，可能有种比较复杂的办法是自己把 BaseRowSerializer 的类不改变package的情况下拷贝出来，然后用 state-processor-API 将相关类强制转换成 RowDataSerializer，不过考虑到你的job graph都是SQL生成的，state-processor-API面向地更多的是data stream

Re: 流跑着跑着，报出了rocksdb层面的错误 Error while retrieving data from RocksDB

Hi， 你可以参阅文档 [1] : 由于 RocksDB 的 JNI API 构建在 byte[] 数据结构之上, 所以每个 key 和 value 最大支持 2^31 字节。 重要信息: RocksDB 合并操作的状态（例如：ListState）累积数据量大小可以超过 2^31 字节，但是会在下一次获取数据时失败。这是当前 RocksDB JNI 的限制。 [1]

Re: 回复：Re: CheckpointedFunction#snapshotState访问键控状态报错

able里面实际kv状态 ---原始邮件--- 发件人:Yun Tang

Re: CheckpointedFunction#snapshotState访问键控状态报错

Hi snapshotState主要是给operator state用的，异常原因是keyed state 访问时需要设置currentKey的，但是currentKey是当前正在处理的record的key，与snapshotState的执行时候的语义不一样，执行snapshotState方法的时候，是可以没有当前record的。 如果想要访问整个keyed state，可以通过 KeyedStateBackend#getKeys(String state, N namespace) 来访问，但还是不建议将keyed

Re: 【疑问】RocksDBStateBackend为什么使用单独封装的frocksdbjni，而不用RocksDB官方提供的RocksJava

] https://issues.apache.org/jira/browse/FLINK-14482 Best Yun Tang From: zoltar9264 Sent: Wednesday, March 24, 2021 13:48 To: user-zh Subject: 【疑问】RocksDBStateBackend为什么使用单独封装的frocksdbjni，而不用RocksDB官方提供的RocksJava 大家好， 在RocksDBStateBackend的pom中看到是使用了 frocksdbjni

Re: procesElement中每天数据都put 进入map，但下一条数据来时map都是空的咨询

Hi 祖安， state抽象的数据结构，无论是value state，list state还是map state，其都是对应流计算处理中的当前key对应的数据结构。以map state具体来说对于每个正在处理的current key （由key selector选择出来 [1]），都有一个对应 的map存储相关的数据，如果你每次都发现对应的map为空，很有可能是因为你的key selector选择出来的key每次都不相同，很大概率是当前处理的record不同导致。 另外，map.isEmpty() 的调用是需要额外开销的（尤其对于RocksDB state

Re: flink升级hadoop3

Hi， Flink自1.11 版本之后就已经支持了hadoop3 [1][2]，具体来讲就是将 HADOOP_CLASSPATH 配置成运行机器上的hadoop3 相关jar包即可。 你也可以参照 [3] 的测试步骤 [1] https://issues.apache.org/jira/browse/FLINK-11086 [2] https://ci.apache.org/projects/flink/flink-docs-stable/deployment/resource-providers/yarn.html#supported-hadoop-versions [3]

Re: 怎么理解 tolerableCheckpointFailureNumber

Hi, tolerableCheckpointFailureNumber 限制的是最大可容忍的连续失败checkpoint计数 continuousFailureCounter [1]，例如将tolerableCheckpointFailureNumber 设置成3，连续失败3次，continuousFailureCounter 会累计到3，作业就会尝试重启。 如果中间有一个checkpoint成功了，continuousFailureCounter 就会重置为零 [2]。

Re: key group from xx to yy does not contain zz异常

Hi， 原因是你的key selector引入了随机变量 （也就是下面的方法keyBy），导致其select出来的key不是固定的 public KeySelector keyBy(int parallelism) { return value -> Joiner.on(SeparatorConstant.BAR).join(value.getMetricsId(), ThreadLocalRandom.current().nextInt(parallelism)); } 例如原先的key selector选出的key是

Re: Flink on yarn JDK 版本支持问题

Hi, MaxMetaspaceSize 是在JDK8中新增的，用以取代以前的PermGen[1]，JDK7中自然不支持。可以在hadoop集群中再安装JDK8，将 env.java.home 指向新的JDK [1] https://www.baeldung.com/java-permgen-metaspace#metaspace 祝好 唐云 From: Jacob <17691150...@163.com> Sent: Saturday, January 23, 2021 16:17 To:

Re: 不同程序间checkpoint迁移

Hi, Flink state processor[1] 应该可以满足你的需求。 [1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/libs/state_processor_api.html 祝好 唐云 From: gimlee Sent: Friday, January 22, 2021 12:07 To: user-zh@flink.apache.org Subject: 不同程序间checkpoint迁移 程序A：jar

Re: Flink sql 状态过期后，checkpoint 大小没变化

文档，开启cleanFullSnapshot是只能对单个状态设置吗，没查到flink sql 开启cleanFullSnapshot的配置的地方？因为只看到StateTtlConfig是对于单个状态的设置，没有对job或者对table的config设置。 2.cleanFullSnapshot 开启后，从checkpoint恢复才会触发清理,不是在checkpoint过程中触发清理掉过期数据？ > 在 2021年1月14日，下午4:48，Yun Tang 写道： > > Hi, > > 你本地的数据肯定是过期了，checkpoint > size没

Re: 回复: 请教个Flink checkpoint的问题

Hi 这是因为在Flink-1.7.0 之后，savepoint也被当做是retained checkpoint了 [1]，当你stop with savepoint 成功时，新的savepoint创建之后，旧的checkpoint因为默认retain checkpoint的数量为1而被subsume掉了，也就是被删掉了。 如果你还想保留之前的一个旧的checkpoint，可以将默认retain的checkpoint数目设置为2 [2]。 另外说一句，即使是已经deprecated的cancel with

Re: Flink sql 状态过期后，checkpoint 大小没变化

backend是rocksdb，没有开启增量，后续没有再插入过数据。 > 在 2021年1月14日，下午4:07，Yun Tang 写道： > > 使用的state backend，以及对应的checkpoint 类型是什么（是否开启incremental checkpoint）？ > > 在一开始插入数据后，直到state TTL超过，期间均没有再插入数据过么？还是说一直在以一定的数据量在插入数据？ > > > 祝好 > 唐云 > > From: 孙啸龙 >

Re: Flink sql 状态过期后，checkpoint 大小没变化

使用的state backend，以及对应的checkpoint 类型是什么（是否开启incremental checkpoint）？ 在一开始插入数据后，直到state TTL超过，期间均没有再插入数据过么？还是说一直在以一定的数据量在插入数据？ 祝好 唐云 From: 孙啸龙 Sent: Thursday, January 14, 2021 15:52 To: user-zh@flink.apache.org Subject: Flink sql 状态过期后，checkpoint 大小没变化 大家好：

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

gt; at > > org.apache.flink.runtime.state.OperatorStateRestoreOperation.restore(OperatorStateRestoreOperation.java:73) > at > > org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder.build(DefaultOperatorStateBackendBuilder.java:83) > ... 15 more > >

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

是有的，难道是访问权限问题吗？B节点无法访问A节点吗，有点奇怪啊，配置了ssh免密的啊，文件夹/data/flink/checkpoints访问权限也设置成了777 Yun Tang 于2021年1月12日周二 下午5:46写道： > Hi > > Flink的容错机制是可以保证TM lost时候会尝试重启作业，“为何任务不能恢复”是需要看完整异常栈的，简单描述是无法帮助排查问题的。 > > 祝好 > 唐云 > > From: Carmen Free > Sent

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

Hi Flink的容错机制是可以保证TM lost时候会尝试重启作业，“为何任务不能恢复”是需要看完整异常栈的，简单描述是无法帮助排查问题的。 祝好 唐云 From: Carmen Free Sent: Tuesday, January 12, 2021 15:52 To: user-zh@flink.apache.org Subject: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？ hi, rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

Re: Flink代码一直报 java.lang.NullPointerException错误，但变量明明初始化了，却仍然有该错误，找不到原因，请诸位指教。谢谢

Hi， 这个错误其实是kryo初始化时候扔出来的。你自定义的类 SockRowV2，WashDetectionSockValue 等，不符合Flink关于pojo的定义，所以会回退到使用kryo进行序列化/反序列化。建议将相关类在kryo上进行注册 [1]。特别地，如果是thrift或者protobuf的类，需要单独注册[2]，更好的方法其实是建议将你们的自定义类修改为满足Flink的POJO类 [3] [1] https://flink.apache.org/news/2020/04/15/flink-serialization-tuning-vol-1.html#kryo

Re: flink编译报错

Hi, 国内网络环境不太好，其实问题是node.js 安装有问题，可以考虑单独安装一下node js和npm，如果还是不行，在不需要webui的前提下，可以编译时候加上profile “-Pskip-webui-build” 来skip掉该部分的编译。 祝好 唐云 From: Ruguo Yu Sent: Tuesday, January 12, 2021 14:00 To: user-zh@flink.apache.org Subject: Re: flink编译报错 试下这个命令 mvn clean

Re: 请教个Flink savepoint的问题

Hi, 没有暴露现成的API，但是可以参照测试用例的写法，给jobGraph设置 savepointRestoreSettings [1]。 [1] https://github.com/apache/flink/blob/ab87cf4ec0260c30194f3d8e75be1f43318aa32a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ResumeCheckpointManuallyITCase.java#L383 祝好 唐云

Re: Flink 1.11.2版本 实时任务运行 报错 is running beyond physical memory limits. Current usage: 25.0 GB of 25 GB physical memory used; 28.3 GB of 52.5 GB virtual memory used. Killing container

Hi, 有可能是堆外内存超用，可以参考最近中文社区的一篇投稿 《详解 Flink 容器化环境下的 OOM Killed》进行修改，建议先增大 jvm-overhead 相关配置 [1] https://mp.weixin.qq.com/s?__biz=MzU3Mzg4OTMyNQ===2247490197=1=b0893a9bf12fbcae76852a156302de95 祝好 唐云 From: Yang Peng Sent: Thursday, January 7, 2021 12:24 To:

Re: Re:Re: Re:flink作业通过grafana监控，若想发出报警该如何选择指标的问题

因为numRestarts 是一个累计值，所以你得区分当前值和之前的数值是否发生了增加，来区分是否发生了failover。 另外，不建议使用YARN的application状态来判断Flink作业状态，因为如果Flink作业配置了重试策略，即使作业不断进行failover，整个YARN的application状态仍然是RUNNING，并不能发现问题。 祝好 唐云 From: bradyMk Sent: Thursday, January 7, 2021 16:38 To:

Re: flink作业通过grafana监控，若想发出报警该如何选择指标的问题

Hi 可以使用 numRestarts [1] 指标进行报警，不过需要维护状态，也就是该值增大时报警。对于旧版本Flink，可以使用以及废弃的fullRestarts 指标。 [1] https://ci.apache.org/projects/flink/flink-docs-stable/ops/metrics.html#availability 祝好 唐云 From: bradyMk Sent: Wednesday, January 6, 2021 18:57 To:

Re: 回复： flink 1.12 Cancel Job内存未释放(问)

Hi 徐州州 请查看一下checkpoint UI部分的overview，观察restored部分的是否为空，也就是没有从checkpoint恢复，同样可以观察job manager 部分日志，看是否从checkpoint resume。 如果没有从checkpoint/savepoint恢复，作业其实相当于是从头重新跑，除非作业有其他的外部访问，否则不应该有任何历史数据能看到。 祝好 唐云 From: 徐州州 <25977...@qq.com> Sent: Tuesday, January 5, 2021 10:34

Re: flink如何使用oss作为checkpoint/savepoint/statebackend?

Hi 其实社区文档 [1] 已经给了很详细的步骤： 1. 将flink-oss-fs-hadoop jar包放在plugins目录下 2. 配置oss的endpoint，id和secret 3. 在需要使用oss的地方，声明oss:// 开头的schema，例如state backend创建的时候 [1] https://ci.apache.org/projects/flink/flink-docs-stable/deployment/filesystems/oss.html 祝好 唐云

Re: Flink RocksDBStateBackend 可以设置阿里云 OSS 存储吗？

Hi 王磊， 当然是可以的，state backend的checkpoint地址其实依赖于Flink的file system实现，只要参照文档[1]的描述，对oss进行相关配置即可。 [1] https://ci.apache.org/projects/flink/flink-docs-stable/deployment/filesystems/oss.html#shaded-hadoop-oss-file-system 祝好 唐云 From: Lei Wang Sent: Monday, December

Re: checkpoint持久化问题

Hi 既然UI上已经显示成功了，一定是成功且成功保存到HDFS上了，可以看下父目录的情况，chk-x 目录可能随着新的checkpoint完成而被删除 祝好 唐云 From: chen310 <1...@163.com> Sent: Friday, December 25, 2020 16:01 To: user-zh@flink.apache.org Subject: checkpoint持久化问题 问题： flink sql中设置了job挂掉后checkpoint保留

Re: flink1.10.1/1.11.1 使用sql 进行group 和 时间窗口 操作后 状态越来越大

Hi @Storm checkpoint的增量模式目前仅对RocksDB生效，这里的增量是指上传新产生的DB sst文件。而RocksDB的全量模式是将DB的有效kv进行序列化写出，除非有大量的数据没有compaction清理掉，否则不可能出现增量checkpoint size无限膨胀，而全量checkpoint正常的问题，你这里的无限膨胀的size范围是多大呢？ 祝好 唐云 From: Storm☀️ Sent: Tuesday, December 22, 2020 19:52 To:

Re: flink1.9.1 如何配置RocksDB的block-cache-usage参数

Hi 这些metrics启用的配置是放到flink conf里面的，不是让你直接在代码里面调用的。 祝好 唐云 From: bradyMk Sent: Thursday, December 17, 2020 20:56 To: user-zh@flink.apache.org Subject: Re: flink1.9.1 如何配置RocksDB的block-cache-usage参数 谢谢 Yun Tang 大佬的解答~ 另外，还想请教一下：我在代码中设置开启了cur-size-all-mem-tables的监

Re: flink1.9.1 如何配置RocksDB的block-cache-usage参数

Hi write buffer的指标可以看 cur-size-all-mem-tables，由于1.9没有block cache的指标，如果不自行将代码[1]pick回去的话，暂时没办法通过内置的方式监控了。 [1] https://issues.apache.org/jira/browse/FLINK-15387 祝好 唐云 From: bradyMk Sent: Wednesday, December 16, 2020 12:03 To: user-zh@flink.apache.org Subject:

Re: 关于flink cdc的N流join状态后端的选择问题： ‎FsStateBackend和‎RocksDBStateBackend

Hi 在前面的邮件里面，已经提示可以使用 async-profiler [1] 来观察RocksDB的内部相关调用栈，这样能看到cpu是否在等待IO。另外iostat等工具可以看磁盘压力如何。至于数据倾斜，可以去看rocksDB的db目录大小，或者看各个subtask的input bytes大小，看看是否task间存在数据倾斜。 [1] https://github.com/jvm-profiling-tools/async-profiler 祝好 唐云 From: jindy_liu

Re: flink-1.11.2 rocksdb when trigger savepoint job fail and restart

Hi 请问你的TM是单slot吧，managed memory是多少？ RocksDB state backend在执行savepoint的时候，会使用一个iterator来遍历数据，所以会存在额外的内存开销（并且该部分开销并不属于write buffer与block cache管理的部分），当然RocksDB的iterator是一个多层的最小堆iterator，理论上来说占用的临时内存并不会太多。不知你们能否将程序抽象成一个必现的demo来给我们做debug呢？ 至于如何解决该问题，可以考虑增大JVM overhead [1] 来增大该部分的buffer空间。 [1]

Re: 关于flink cdc的N流join状态后端的选择问题： ‎FsStateBackend和‎RocksDBStateBackend

Hi 首先需要纠正一点的是，RocksDB的实际可用内存并不是你以为的13GB，因为从Flink-1.10 开始引入的 managed memory [1][2]，会将slot上的RocksDB的实际可用内存限制在 managed memory / number of slots，也就是说对于你配置的10个slot，20GB的process memory，0.75的managed fraction，真实的per slot managed memory其实只有不到1.5GB，也就是说你配置的write buffer count以及max write

Re: 回复：flink使用RocksDB增量checkpoints，程序运行一段时间报出：超出物理内存

Hi Operator state 本身也并不是线程安全的，只是往常的读写都是持有checkpoint锁的task主线程或者checkpoint异步线程，所以才能做到数据安全，SourceFunction文档里面也强调需要在获得checkpointLock的前提下更新state。 至于如何开启Flink中的RocksDB的native metrics，之前给你的文档链接里面有描述，相关的配置项设为true即可。 祝好 唐云 From: bradyMk Sent: Thursday, December 10,

Re: 关于flink cdc的N流join状态后端的选择问题： ‎FsStateBackend和‎RocksDBStateBackend

: Thursday, December 10, 2020 11:04 To: user-zh Cc: Yun Tang Subject: Re: 关于flink cdc的N流join状态后端的选择问题： ‎FsStateBackend和‎RocksDBStateBackend 关于 rocksdb 的性能调优, @Yun Tang<mailto:myas...@live.com> 会更清楚。 On Thu, 10 Dec 2020 at 11:04, Jark Wu mailto:imj...@gmail.com>> wrote: 建议大状态还是用 rock

Re: 回复：flink使用RocksDB增量checkpoints，程序运行一段时间报出：超出物理内存

State本身不是线程安全的 [1]，但是目前对于state的更新都是在task主线程内，而task主线程是线程安全的。除非通过一些特别的方式，例如异步的metrics线程用户逻辑下访问state导致的state写更新副作用，一般是不会出现写错的问题。 [1] https://issues.apache.org/jira/browse/FLINK-13072 祝好 唐云 From: bradyMk Sent: Tuesday, December 8, 2020 17:59 To:

Re: 答复: flink使用RocksDB增量checkpoints，程序运行一段时间报出：超出物理内存

配置一下 RocksDB 的native metrics，看下block cache以及 write buffer的实际使用内存。 另外，Flink中一个state会使用一个RocksDB的column family，而write buffer和block cache是一套column family 一套，所以你的operator 内的state数目多，slot内的keyed operator多，都会导致内存成倍增长 [1]

Re: 答复: flink使用RocksDB增量checkpoints，程序运行一段时间报出：超出物理内存

RocksDB只是将数据可以存储在磁盘上，Flink再周期性将磁盘上数据上传到HDFS，内存中还是有LSM的write buffer以及block cache，也还是需要使用内存的 建议升级Flink版本到1.10+，引入了managed memory功能，理论上对于内存控制是要好很多的。 祝好 唐云 From: bradyMk Sent: Monday, December 7, 2020 11:27 To: user-zh@flink.apache.org Subject: Re: 答复:

Re: Flink任务启动偶尔报错PartitionNotFoundException，会自动恢复。

Hi 集群负载比较大的时候，下游一直收不到request的partition，就会导致PartitionNotFoundException，建议增大 taskmanager.network.request-backoff.max [1][2] 以增大重试次数 [1] https://ci.apache.org/projects/flink/flink-docs-stable/ops/config.html#taskmanager-network-request-backoff-max [2]

Re: Re:flink内存超用问题

Hi 可以通过增大 "taskmanager.memory.jvm-overhead.max" [1] 以及 "taskmanager.memory.process.size" [2] 来增大可以超用的内存空间。可以通过观察 "state.backend.rocksdb.metrics.block-cache-pinned-usage" [3] 的数值看rocksDB使用的native memory是否超过managed memory。 [1]

Re: Checkpoint size的问题

Hi web UI显示的是增量上传数据量，包括各个task上传的数据，而_metadata 只是一个元数据，是由JM上传的，所以不能将_metadata与checkpoint UI显示的数据量划等号。 祝好 唐云 From: gsralex Sent: Wednesday, October 28, 2020 19:17 To: user-zh@flink.apache.org Subject: Checkpoint size的问题 Hi, All Checkpoint 一般Web

Re: 回复：请问现在Flink支持动态扩缩容吗？

Hi Flink-1.8 之前支持通过rest命令进行扩缩容 [1]，不过后来在重构时该功能被disable了[2]。当然这个功能距离动态扩缩容还是有差距的，可以理解成是从外部进行扩缩容的基础。 目前在阿里巴巴的企业版中，有名为libra的动态扩缩容插件 [3] 提供相关功能。 [1] https://ci.apache.org/projects/flink/flink-docs-release-1.8/monitoring/rest_api.html#jobs-jobid-rescaling [2]

Re: flink-windows-state

Hi 这里涉及到的问题比较多。 1. 为什么心跳会超时？是因为Full GC么，如果是使用的FsStateBackend/MemoryStateBackend，这是比较好解释的，因为数据在JVM堆上。如果使用的是RocksDB，这里是解释不通的。 2. window确实是使用state来存储数据，如果认为自己的state太大的话，是不是因为使用不当呢？可以参考文档 [1] 进行调优 3. 仍在运行的TM里面在做什么呢，为什么没有被JM释放，需要检查相关孤儿TM的日志以及jstack查看进程操作判断。 [1]

Re: 回复：rocksdb增量ckeckpoint问题

Hi，云昆 首先，如果Congxian回答的，sst文件是不变的，所以就有了“增量”的前提，如果多个checkpoint均包含某个sst文件，那么该文件可以在多个checkpoint间共享，这些文件表示是属于shared的文件，存储在shared目录下，可以参考文档[1]。

Re: flink1.11.2 在k8s上部署,如何启动history server

Hi 可以在yaml文件中覆盖原始的 ENTRYPOINT 启动命令 [1] 或者可以参考 FLINK-17167 [2] 中的修改更改原始Dockerfile中的docker-entrypoint.sh [1] https://kubernetes.io/zh/docs/tasks/inject-data-application/define-command-argument-container/ [2] https://issues.apache.org/jira/browse/FLINK-17167 祝好 唐云

Re: flink savepoint和checkpoint相关事项

Hi 在修改作业的情况下，从checkpoint恢复的主要措施是确保相关operator设置了uid[1]，恢复state的时候添加了允许non-restored state [2] 恢复 [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/state/savepoints.html#assigning-operator-ids [2]

Re: checkpoint问题

Hi checkpoint使用operator id进行一一映射进行恢复，请参照 设置id[1]，以及如果checkpoint中存在某个operator但是修改后的作业并不存在该operator时的处理逻辑[2] [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/state/savepoints.html#assigning-operator-ids [2]

Re: 从Savepoint/Checkpoint恢复时 keyedstate中部分数据无法被读取

Hi 首先，transient ListState 这种用法绝对是没有问题的，你可以在Flink源码中找到大量的使用例子 [1] 可以排查的思路 1. 你的state是否开启了TTL呢 2. 能否在写入的时候，进行打印，然后在get会空iterator时进行对比，确认之前是写入成功的。 3. 目前使用的state backend是什么，更换一种state backend，问题还能复现么 [1]

Re: flink on k8s 如果jobmanager 所在pod重启后job失败如何处理

jobmanager high availability[1] and you could refer to [2] for plans of HighAvailabilityService based on native k8s APIs. [1] https://ci.apache.org/projects/flink/flink-docs-stable/ops/jobmanager_high_availability.html [2] https://issues.apache.org/jira/browse/FLINK-12884 Best Yun Tang

Re: Flink如何实现至多一次（At Most Once）

Hi 如果是完全依赖source的offset管理，可以达到类似 at most once 的语义。 社区其实也有更完备的checkpoint at most once 的实现讨论，已经抄送了相关的开发人员 @yuanmei.w...@gmail.com 祝好 唐云 From: Paul Lam Sent: Thursday, September 3, 2020 17:28 To: user-zh Subject: Re:

Re: 从Savepoint/Checkpoint恢复时 keyedstate中部分数据无法被读取

Hi 我觉得这个不是root cause，实际上 transient ListState 是一种正确的用法，因为state应该是在函数open方法里面进行初始化，所以transient 修饰即可。 麻烦把这个list state的初始化以及使用方法的代码都贴出来吧。 祝好 唐云 From: Liu Rising Sent: Thursday, September 3, 2020 12:26 To: user-zh@flink.apache.org Subject: Re:

Re: flink文档

Hi SQL解析不通过的可以在 https://issues.apache.org/jira/projects/FLINK/issues 里面创建相关ticket指明出来，很快会有相关开发来帮助的。 不过需要注意的是，需要用英文进行阐述。 祝好 唐云 From: Dream-底限 Sent: Friday, August 28, 2020 16:42 To: user-zh@flink.apache.org Subject: flink文档 hi、

Re: flink prometheus 无法上报accumulator类型监控吗

Hi 没有名为 accumulator 的metrics类型数据，目前只有Counters, Gauges, Histograms 和 Meters [1] 这四种，如果你想要用累积型metrics，可以考虑counters [1] https://ci.apache.org/projects/flink/flink-docs-stable/monitoring/metrics.html#metric-types 祝好 唐云 From: 赵一旦 Sent: Friday, August 28, 2020

Re: 从savepoint 启动以后，无法在checkpoint页面看到last restore的相关信息

Hi 范超 虽然看不到你的图，但是你的启动命令错误了，所有的options应该放在jar包文件地址前面[1] 1. class name 应该在 jar包地址前面 [2] 2. savepoint/checkpoint 地址应该在jar包地址前面 [3] 没有正确从checkpoint恢复的原因应该是这个原因 [1] https://ci.apache.org/projects/flink/flink-docs-stable/ops/cli.html#usage [2]

Re: [ANNOUNCE] New PMC member: Dian Fu

Congratulations , Dian! Best Yun Tang From: Yang Wang Sent: Friday, August 28, 2020 10:28 To: Arvid Heise Cc: Benchao Li ; dev ; user-zh ; Dian Fu ; user Subject: Re: [ANNOUNCE] New PMC member: Dian Fu Congratulations Dian ! Best, Yang Arvid Heise

Re: 回复： 流处理任务中checkpoint失败

Hi Robert 你的两个source firstSource和secondSource是自己实现的么，一种怀疑是source在没有数据时持锁[1][2]导致checkpoint barrier并没有下发。 建议使用jstack查看在没有数据下发时，source相关task的java调用栈，观察是否存在等待锁释放 [1]

Re: 1.11.2大概什么时候发布

可以参照 https://flink.apache.org/downloads.html#all-stable-releases 的历史发布记录，一般是3个月左右，也就是大约10月底。 1.11.2 有什么特别期待的bug fix么？ 祝好 唐云 From: abc15...@163.com Sent: Wednesday, August 26, 2020 15:41 To: user-zh@flink.apache.org Subject: 1.11.2大概什么时候发布 1.11.2大概什么时候发布？

Re: flink checkpoint导致反压严重

Hi 对于已经改为at least once的checkpoint，其在checkpoint时对于作业吞吐的影响只有task 同步阶段的snapshot，这个时间段的snapshot由于与task的主线程的数据访问持有同一把锁，会影响主线程的数据处理。但是就算这样，我也很怀疑checkpoint本身并不是导致早上10点高峰期无法运行的罪魁祸首。 使用异步的，支持增量的state backend （例如RocksDBStateBackend）会大大缓解该问题。 建议排查思路： 1. 检查使用的state backend类型 2.

Re: [ANNOUNCE] Apache Flink 1.10.2 released

Thanks for Zhu's work to manage this release and everyone who contributed to this! Best, Yun Tang From: Yangze Guo Sent: Tuesday, August 25, 2020 14:47 To: Dian Fu Cc: Zhu Zhu ; dev ; user ; user-zh Subject: Re: [ANNOUNCE] Apache Flink 1.10.2 released Thanks

Re: 有没有可能使用tikv作为flink 分布式的backend

Hi 这种思路我觉得是可以尝试的，不过目前看需要改动的地方很多： 1. 需要更改RocksDB 创建checkpoint 到TiKV的代码逻辑 2. 需要改动RocksDB 从checkpoint resume的代码逻辑 3. 如果想要数据可以TiKV可以读取，那么TiKV中存储的格式要么与RocksDB内存储的一样，那这样子的话，lookup时候，需要能够反序列化Flink在RocksDB中的存储格式；要么是重新的格式，但这样子会导致RocksDB的checkpoint流程和时间都会增长。 4. TiKV中的数据的更新依赖于checkpoint

Re: 有没有可能使用tikv作为flink 分布式的backend

Hi TiKV 本身就是分布式的，多副本的，可以类比HBase，所以不是将其向Flink内置的state backend靠拢，而是向Flink读写HBase靠拢，这样若干写TiKV的Flink作业就做到了数据共享。 如果想将TiKV向Flink state-backend靠拢，TiKV本身的分布式架构，多副本机制，网络传输（而不是本地磁盘访问）都是缺点或者说不再必要存在的特性。 最后就会演化成现在Flink + RocksDB state-backend的架构，更何况TiKV就是基于RocksDB的，整体意义不是很大。 祝好 唐云

Re: state序列化问题

要想知道，在MapStateDescriptor声明类型信息时，我是否应该把内部Map声明成明确的HashMap类型，而不是Map类型？ Yun Tang 于2020年8月21日周五 上午12:13写道： > Hi > > 如果想要在list中保存String，也就是list中的每个元素格式是String，ListState的格式应该是 > ListState, 而不是 > ListState>，后者表示有一个list，list中的每一个元素均是一个list > > ListState 本身并不属于java的collection，所以不存在ArrayLis

Re: state序列化问题

Hi 如果想要在list中保存String，也就是list中的每个元素格式是String，ListState的格式应该是 ListState, 而不是 ListState>，后者表示有一个list，list中的每一个元素均是一个list ListState 本身并不属于java的collection，所以不存在ArrayList 与 LinkedList的区别。 祝好 唐云 From: shizk233 Sent: Thursday, August 20, 2020 18:00 To:

Re: 增量che ckpoint

Hi 增量checkpoint与web界面的信息其实没有直接联系，增量checkpoint的信息记录由CheckpointCoordinator中的SharedStateRegistry[1] 进行计数管理，而保留多少checkpoint则由 CheckpointStore管理 [2]. 保留2个checkpoint的执行过程如下： chk-1 completed --> register chk-1 in state registry --> add to checkpoint store chk-2 completed --> register chk-2 in state

Re: 能否考虑针对检查点和保存点设置不同的超时时间

Hi 你的这个需求其实社区早已经有相关ticket [1]了，不过这个需求一直不是很强烈，毕竟大多数时候可以通过增大checkpoint timeout即可，增大checkpoint timeout不代表着也会增大checkpoint占据的资源。 [1] https://issues.apache.org/jira/browse/FLINK-9465 祝好 唐云 From: 赵一旦 Sent: Tuesday, August 18, 2020 14:38 To: user-zh@flink.apache.org

Re: flink 1.11 web ui请教

Hi 1. 框图的数量是因为默认启用了operator chain导致的，至于连接线上的文字（例如hash）则是由网络连接方式决定了[2] 2. record received 为0 是因为这个指标表征了数据在Flink 的channel内收到的record数量，由于source节点并没有从Flink channel获取数据（往往是从外部系统获取），所以自然record received为0 [1]

  1   2   >