Re: 采集mysql全量的时候出现oom问题
从报错信息看，是由于JM的堆内存不够，可以尝试把JM内存调大，一种可能的原因是mysql表全量阶段分片较多，导致SourceEnumerator状态较大。 Best, Shawn Huang wyk 于2024年4月8日周一 17:46写道： > > > 开发者们好: > flink版本1.14.5 > flink-cdc版本 2.2.0 > > 在使用flink-cdc-mysql采集全量的时候,全量阶段会做checkpoint,但是checkpoint的时候会出现oom问题,这个有什么办法吗? >具体报错如附件文本以及下图所示: > > >
Re: flink cdc metrics 问题
你好，目前flink cdc没有提供未消费binlog数据条数这样的指标，你可以通过 currentFetchEventTimeLag 这个指标（表示消费到的binlog数据中时间与当前时间延迟）来判断当前消费情况。 [1] https://github.com/apache/flink-cdc/blob/master/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/metrics/MySqlSourceReaderMetrics.java Best, Shawn Huang casel.chen 于2024年4月8日周一 12:01写道： > 请问flink cdc对外有暴露一些监控metrics么？ > 我希望能够监控到使用flink cdc的实时作业当前未消费的binlog数据条数，类似于kafka topic消费积压监控。 > 想通过这个监控防止flink cdc实时作业消费慢而被套圈（最大binlog条数如何获取？）
Re: Re: 1.19自定义数据源
你好，关于如何实现source接口可以参考以下资料： [1] FLIP-27: Refactor Source Interface - Apache Flink - Apache Software Foundation <https://cwiki.apache.org/confluence/display/FLINK/FLIP-27%3A+Refactor+Source+Interface> [2] 如何高效接入 Flink：Connecter / Catalog API 核心设计与社区进展 (qq.com) <https://mp.weixin.qq.com/s/CAxAgVDzHZ6vZ3BSvi9mlA> Best, Shawn Huang liuchao 于2024年3月28日周四 15:39写道： > 找一个实现source接口的算子，参考一下 > > > 刘超 > liuchao1...@foxmail.com > > > > > > > > > --原始邮件-- > 发件人: > "user-zh" > < > ha.fen...@aisino.com; > 发送时间:2024年3月28日(星期四) 下午3:34 > 收件人:"user-zh" > 主题:Re: Re: 1.19自定义数据源 > > > > 我想问的就是如果需要实现Source接口，应该怎么写，有没有具体的例子实现一个按照一定速度生成自定义的类？ > > 发件人： gongzhongqiang > 发送时间： 2024-03-28 15:05 > 收件人： user-zh > 主题： Re: 1.19自定义数据源 > 你好： > > 当前 flink 1.19 版本只是标识为过时，在未来版本会移除 SourceFunction。所以对于你的应用而言为了支持长期 flink > 版本考虑，可以将这些SourceFunction用Source重新实现。 > > ha.fen...@aisino.com > > > 原来是继承SourceFunction实现一些简单的自动生成数据的方法，在1.19中已经标识为过期，好像是使用Source接口，这个和原来的SourceFunction完全不同，应该怎么样生成测试使用的自定义数据源呢？ >
Re: 退订
Hi，退订可以发送任意内容的邮件到 user-zh-unsubscr...@flink.apache.org 来取消订阅来自 user-zh@flink.apache.org 邮件列表的邮件，邮件列表的订阅管理，可以参考[1] [1] https://flink.apache.org/zh/what-is-flink/community/ Best, Shawn Huang 雷刚 于2024年2月29日周四 14:41写道： > 退订
Re: flink sql作业如何统计端到端延迟
Flink有一个端到端延迟的指标，可以参考以下文档[1]，看看是否有帮助。 [1] https://nightlies.apache.org/flink/flink-docs-release-1.18/zh/docs/ops/metrics/#end-to-end-latency-tracking Best, Shawn Huang casel.chen 于2024年2月21日周三 15:31写道： > flink sql作业从kafka消费mysql过来的canal > json消息，经过复杂处理后写入doris，请问如何统计doris表记录的端到端时延？mysql表有update_time字段代表业务更新记录时间。 > doris系统可以在表schema新增一个更新时间列ingest_time，所以在doris表上可以通过ingest_time - > update_time算出端到端时延，但这种方法只能离线统计，有没有实时统计以方便实时监控的方法？ > > 查了SinkFunction类的invoke方法虽然带有Context类型参数可以获取当前处理时间和事件时间，但因为大部分sink都是采用攒微批方式再批量写入的，所以这两个时间直接相减得到的时间差并不能代表真实落库的时延。有没有精确获取时延的方法呢？
Re: 退订
Hi，退订可以发送任意内容的邮件到 user-zh-unsubscr...@flink.apache.org 来取消订阅来自 user-zh@flink.apache.org 邮件列表的邮件，邮件列表的订阅管理，可以参考[1] [1] https://flink.apache.org/zh/what-is-flink/community/ Best, Shawn Huang 18679131354 <18679131...@163.com> 于2024年2月27日周二 14:32写道： > 退订
Re: flink job task在taskmanager上分布不均衡
看你的描述应该是Standalone部署模式。 默认调度方法是以slot为单位的，并且会倾向于分配到同一个TaskManager的slot中。 想要充分利用所有slot，一个方法是把集群中slot总数设为所有作业的并行度之和， 或者尝试将配置项cluster.evenly-spread-out-slots设为true。 Best, Shawn Huang 张锴 于2021年5月7日周五 下午7:50写道： > 给l另一个job设置个组别名，不同的组不会slot共享，会跑到别的slot上，slot可以灵活的运行在不同的TM上。 > > allanqinjy 于2021年5月7日周五 下午7:38写道： > > > > > > https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/config.html > > flink的配置中是有flink taskmanager配置的，一个tm对应几个slots > > > 。taskmanager.numberOfTaskSlots默认是1.并行度是对应了slots数据，一般我们的slots与并行度最大的一样。你可以看一下这个参数设置。然后对照官网说明。 > > > > > > | | > > allanqinjy > > | > > | > > allanqi...@163.com > > | > > 签名由网易邮箱大师定制 > > > > > > 在2021年05月7日 16:42，wenyuan138 写道： > > flink集群(flink 1.10.1)，taskmanager有4个，每个有10个slot。 然后我有2个job， > > 每个并行度是4，预期是会分布到不同taskmanager的slot上(也就是4个taskmanager平均分配2个slot， > > 这样能更好的利用cpu资源). 结果发现这2个job的8个task全部分配到同一个taskmanager上了。 为什么？ > > 有什么配置可以改变这种行为吗？ > > 我们想要的是task能分到不同的taskmanager上。 谢谢！ > > > > > > > > -- > > Sent from: http://apache-flink.147419.n8.nabble.com/ >
Re: 退订
Hi, 退订需要发邮件到 user-zh-unsubscr...@flink.apache.org 可以参考 https://flink.apache.org/zh/community.html#section-1 Best, Shawn Huang Natasha <13631230...@163.com> 于2021年1月22日周五 下午5:04写道： > > > 退订
Re: 帮忙推荐下flink是用啥可视化的客户端？
你说的客户端是指什么？Flink 默认在 8081 端口提供了 Web UI，可以提交和取消任务，查看日志和一些基础指标。 Best, Shawn Huang yinghua...@163.com 于2020年12月3日周四 下午8:46写道： >
Re: 邮件退订
Hi, 退订需要发邮件到 user-zh-unsubscr...@flink.apache.org 可以参考 https://flink.apache.org/zh/community.html#section-1 Best, Shawn Huang wangleigis 于2020年11月14日周六 上午11:55写道： > > > > > > > 退订 > > > > > > > > > > > -- > > 祝：工作顺利，完事如意！
Re: slot数量与并行度的大小关系
Hi， Flink 的调度策略会保证一个job需要的slot数恰好等于该job所有算子的最大并行度。 如果slot数量小于算子的最大并行度，则该job无法执行。可以参考[1][2]中的文档描述。 目前没有方法让flink自动选择可用slot数量作为并行度，但可以通过[3]中的几种方法来设置。 [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/internals/job_scheduling.html [2] https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/concepts/flink-architecture.html#task-slots-and-resources [3] https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/parallel.html Best, Shawn Huang hl9...@126.com 于2020年11月11日周三 下午2:58写道： > Hi，all： > 我在flink > web面板上提交了1个job，job的并行度为15，flink集群slot总数为12，发现任务一直在created阶段等待，一段时间后报错： > Caused by: > org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: > Could not allocate the required slot within slot request timeout. > Please make sure that the cluster has enough resources. > > 是因为slot数量必须要大于并行度吗？有没有参数可以让flink自动选择可用slot数作为job的并行度？ > > > > hl9...@126.com >
Re: 关于cluster.evenly-spread-out-slots参数的底层原理
我说一下我看源码（1.11.2）之后的理解吧，不一定准确，仅供参考。 cluster.evenly-spread-out-slots 这个参数设置后会作用在两个地方： 1. JobMaster 的 Scheduler 组件 2. ResourceManager 的 SlotManager 组件 对于 JobMaster 中的 Scheduler， 它在给 execution vertex 分配 slot 是按拓扑排序的顺序依次进行的。 Scheduler 策略是会倾向于把 execution vertex 分配到它的上游节点所分配到的slot上， 因此在给某个具体 execution vertex 分配 slot 时都会计算出一个当前节点倾向于选择的TaskManager集合， 然后在可选的 slot 候选集中会根据三个维度来为某个slot打分，分别是： 1. 候选slot所在的 TaskManager 与倾向于选择的 TaskManager 集合中有多少个的 ResourceID 是相同的（对于standalone模式可以不考虑该维度） 2. 候选slot所在的 TaskManager 与倾向于选择的 TaskManager 集合中有多少个的 全限定域名 是相同的 3. 候选slot所在的 TaskManager 目前的资源占用率 只有配置了 cluster.evenly-spread-out-slots 后，才会考虑第三个维度，否则仅会用前面两个维度进行打分。 打分之后会选择得分最高的 slot 分配给当前的 exection vertex。 需要注意的是这里的资源利用率只是根据某个 slot 所在的 TaskManager 中剩下多少个能够分配该 execution vertex 的 slot 计算出的， （因为 Flink 要求同一 job vertex 的并行任务不能分配到同一 slot 中），能分配的越多，资源利用率越小，否则利用率越大。 而不是指实际的CPU内存等资源利用率。 对于 ResourceManager 中的 SlotManager 组件（这里说的都是 Standalone 模式下的 ResourceManager）， 由于 JobMaster 的 slot 都是要向 resource manager 申请的。如果 JobMaster 需要新的 slot 了，会向 ResourceManager 的 SlotManager 组件申请。 如果没有配置 cluster.evenly-spread-out-slots 的话，SlotManager 从可用 slot 中随机返回一个。 如果配置了 cluster.evenly-spread-out-slots，SlotManager 会返回资源利用率最小的一个 slot。 这里的资源利用率计算方式是：看某个 slot 所在的 TaskManager 中有多少 slot 还没有被分配，空闲的越多，利用率越小，否则越大。 最后，你提问中说的均衡我没有太理解。某个算子的并发子任务是不会被分配到同一个slot中的， 但如果想把这些子任务均匀分配到不同机器上，这个当前的调度算法应该是无法保证的。 Best, Shawn Huang 赵一旦 于2020年11月5日周四 下午10:18写道： > 有没有人对cluster.evenly-spread-out-slots参数了解比较深入的给讲解下。 > > 我主要想知道，设置这个参数为true之后。Flink是以一个什么样的规则去尽可能均衡分配的。 > standalone集群模式下，每个机器性能相同，flink slot数量配置相同情况下。基于*这种分配规则*，有没有一种方法让Flink做到 > *完全均衡*，而*不是尽可能均衡*？ > > 此外，我说的“均衡”都特指算子级别的均衡。不要5机器一共5个slot，然后任务有5个算子，每个算子单并发并且通过不同的share > group各独占1个slot这种均衡。我指的是每个算子都均衡到机器（*假设并发设置合理*）。 >
