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这种均衡。我指的是每个算子都均衡到机器(*假设并发设置合理*)。 >