Re: flink on yarn 模式，在jar任务中，怎么获取rest port

在 flink-conf.yaml 中可以指定 rest.port， 可指定一个范围 On Wed, Jul 31, 2024 at 8:44 PM melin li wrote: > flink on yarn 模式, rest port 是随机的，需要获取rest port，有什么好办法？ >

Re: flink 任务运行抛ClassNotFoundException

看了下没问题。而且，这个 > 'org.apache.hudi.com.xx.xx.xxx.A' 在我的任务 jar 里面确实是存在的 > > > > > --原始邮件-- > 发件人: > "user-zh" > < > decq12y...@gmail.com

Re: flink 任务运行抛ClassNotFoundException

你好，这个类被 shade 了，但是调用这个类的其他类可能在不同的 jar 包，没有都被 shade 处理。可以 grep -rn 'org.apache.hudi.com.xx.xx.xxx.A' 看看所有调用这个类的包是不是都做了 shade 处理。 ℡小新的蜡笔不见嘞、 <1515827...@qq.com.invalid> 于2024年7月18日周四 18:31写道： > 请问，Flink 任务运行期间 偶尔会抛出 ClassNotFoundException 异常，这个一般是什么原因，以及怎么解决呢？信息如下： > * 这个类确实存在于 任务Jar 里面 > * 这个类是经过

Re: Flink Standalone-ZK-HA模式下，CLi任务提交

从日志看，ZK 集群滚动的时候发生了切主，两个 JM 都先后成为过 Leader，但是并没有同时是 Leader。 Best, Zhanghao Chen From: love_h1...@126.com Sent: Friday, July 12, 2024 17:17 To: user-zh@flink.apache.org Subject: Flink Standalone-ZK-HA模式下，CLi任务提交 版本：Flink 1.11.6版本，Standalone HA模式，ZooKeeper 3.5.8版本

Re: Flink如何做到动态关联join多张维度表中的n张表？

> > > > > --原始邮件-- > 发件人: > "user-zh" > < > xhzhang...@gmail.com; > 发送时间:2024年6月19日(星期三) 下午5:55 > 收件人:"user-zh&q

Re: Flink如何做到动态关联join多张维度表中的n张表？

维表更新后要刷新历史的事实表吗？这个用flink来做的话，几乎不太可能实现，尤其是涉及到多个维表，相当于每次维表又更新了，就要从整个历史数据里面找到关联的数据，重新写入。不管是状态存储，还是更新数据量，需要的资源都太高，无法处理。 在我们目前的实时宽表应用里面，实时表部分一般都是流水类的，取到的维表信息，就应该是业务事实发生时的数据。 维表更新后刷新事实的，一般都是夜间批量再更新。如果有强实时更新需求的，只能在查询时再关联维表取最新值 王旭 于2024年6月16日周日 21:20写道： > 互相交流哈，我们也在做类似的改造 >

Re: flink cdc 3.0 schema变更问题

你好，DataStream 的方式需要设置 includeSchemaChanges(true) 参数，并且设置自定义的 deserializer，参考这个链接[1]。 如果不想使用 json 的方式，希望自定义 deserializer，从 SourceRecord 里提取 ddl 的方式可以参考这个链接[2]提供的方案。 [1]

Re: flink cdc 3.0 schema变更问题

Zapjone 好， 目前的 Schema Evolution 的实现依赖传递 CDC Event 事件的 Pipeline 连接器和框架。如果您希望插入自定义算子逻辑，建议参考 flink-cdc-composer 模块中的 FlinkPipelineComposer 类构建算子链作业的方式，并在其中插入自定义的 Operator 以实现您的业务逻辑。 另外，对于一些简单的处理逻辑，如果能够使用 YAML 作业的 Route（路由）、Transform（变换）功能表述的话，直接编写对应的 YAML 规则会更简单。 祝好！ Regards, yux De : zapjone

Re: Flink 1.18.1 ，重启状态恢复

看起来和 FLINK-34063 / FLINK-33863 是同样的问题，您可以升级到1.18.2 试试看。 [1] https://issues.apache.org/jira/browse/FLINK-33863 [2] https://issues.apache.org/jira/browse/FLINK-34063 陈叶超 于2024年5月16日周四 16:38写道： > > 升级到 flink 1.18.1 ，任务重启状态恢复的话，遇到如下报错： > 2024-04-09 13:03:48 > java.lang.Exception: Exception while

Re: Flink sql retract to append

以处理时间为升序，处理结果肯定不会出现回撤的，因为往后的时间不会比当前时间小了，你可以在试试这个去重 On Tue, Apr 30, 2024 at 3:35 PM 焦童 wrote: > 谢谢你的建议 但是top-1也会产生回撤信息 > > > 2024年4月30日 15:27，ha.fen...@aisino.com 写道： > > > > 可以参考这个 > > > https://nightlies.apache.org/flink/flink-docs-release-1.19/zh/docs/dev/table/sql/queries/deduplication/ > >

Re: Flink sql retract to append

谢谢你的建议 但是top-1也会产生回撤信息 > 2024年4月30日 15:27，ha.fen...@aisino.com 写道： > > 可以参考这个 > https://nightlies.apache.org/flink/flink-docs-release-1.19/zh/docs/dev/table/sql/queries/deduplication/ > 1.11版本不知道是不是支持 > > From: 焦童 > Date: 2024-04-30 11:25 > To: user-zh > Subject: Flink sql retract to append

Re: Flink流批一体应用在实时数仓数据核对场景下有哪些注意事项？

流模式和批模式在watermark和一些算子语义等方面上有一些不同，但没看到Join和Window算子上有什么差异，这方面应该在batch mode下应该是支持的。具体的两种模式的比较可以看一下这个文档 https://nightlies.apache.org/flink/flink-docs-master/zh/docs/dev/datastream/execution_mode/ On Thu, Apr 18, 2024 at 9:44 AM casel.chen wrote: > > 有人尝试这么实践过么？可以给一些建议么？谢谢！ > > > > > > > > > > > >

Re: flink 已完成job等一段时间会消失

你好： 如果想长期保留已完成的任务，推荐使用 History Server : https://nightlies.apache.org/flink/flink-docs-release-1.19/docs/deployment/config/#history-server Best, Zhongqiang Gong ha.fen...@aisino.com 于2024年4月9日周二 10:39写道： > 在WEBUI里面，已完成的任务会在completed jobs里面能够看到，过了一会再进去看数据就没有了，是有什么配置自动删除吗？ >

Re: flink cdc metrics 问题

你好，目前flink cdc没有提供未消费binlog数据条数这样的指标，你可以通过 currentFetchEventTimeLag 这个指标（表示消费到的binlog数据中时间与当前时间延迟）来判断当前消费情况。 [1]

Re: flink operator 高可用任务偶发性报错unable to update ConfigMapLock

这种一般是因为APIServer那边有问题导致单次的ConfigMap renew lease annotation的操作失败，Flink默认会重试的 如果你发现因为这个SocketTimeoutException原因导致了任务Failover，可以把下面两个参数调大 high-availability.kubernetes.leader-election.lease-duration: 60s high-availability.kubernetes.leader-election.renew-deadline: 60s Best, Yang On Tue, Mar 12,

Re: flink写kafka时，并行度和分区数的设置问题

你好， 写 Kafka 分区的策略取决于使用的 Kafka Sink 的 Partitioner [1]，默认使用的是 Kafka 的 Default Partitioner，底层使用了一种称之为黏性分区的策略：对于指定 key 的数据按照对 key hash 的方式选择分区写入，对于未指定 key 的数据则随机选择一个分区，然后“黏住”这个分区一段时间以提升攒批效果，然后攒批结束写完后再随机换一个分区，来在攒批效果和均匀写入间做一个平衡。 具体可以参考 [2]。 因此，默认配置下不存在你说的遍历导致攒批效果下降的问题，在达到 Kafka

Re: flink集群如何将日志直接写入elasticsearch中？

比较简单的方式是启动一个filebeat进程，抓取 jobmanager.log 和t askmanager.log Best, Jiabao kellygeorg...@163.com 于2024年3月13日周三 15:30写道： > 有没有比较方便快捷的解决方案？ > > >

Re: flink sql关联维表在lookup执行计划中的关联条件问题

Hi iasiuide, 方便share一下你使用的flink版本与jdbc connector的版本吗？据我所了解，jdbc connector在FLINK-33365[1]解决了lookup join条件丢失的相关问题。 [1] https://issues.apache.org/jira/browse/FLINK-33365 祝好~ > 2024年3月8日 11:02，iasiuide 写道： > > > > > 图片可能加载不出来，下面是图片中的sql片段 > .. > END AS trans_type, > >

Re: Re:RE: RE: flink cdc动态加表不生效

Hi， casel chan, 社区已经对增量框架实现动态加表（https://github.com/apache/flink-cdc/pull/3024 ），预计3.1对mongodb和postgres暴露出来，但是Oracle和Sqlserver目前并没暴露，你可以去社区参照这两个框架，将参数打开，并且测试和适配。 Best, Hongshun

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 >

Re: flink cdc底层的debezium是如何注册schema到confluent schema registry的？

Hi，casel.chen。 这个部分应该是在 CDC 项目里没有涉及到，CDC 依赖 debezium 的 engine 部分直接读取出变更数据，并没有像 debezium 本身一样去写入到 Kafka 中。 可以考虑去 Debezium 社区咨询一下这部分的内容，Debezium开发者们应该更熟悉这部分的内容。 祝好， Hang casel.chen 于2024年2月29日周四 18:11写道： > 搜索了debezium源码但没有发现哪里有调用 > SchemaRegistryClient.register方法的地方，请问它是如何注册schema到confluent

Re: flink重启机制

图片没有显示出来。container 调度是由 yarn 控制的，yarn 会优先选择运行中的节点。按理说 container 不会调度到下线的节点，你通过 yarn web 或者 yarn node -list 确认了吗？ chenyu_opensource 于2024年2月27日周二 18:30写道： > 你好，flink任务提交到yarn上，由于某个节点下线导致flink任务失败，如下： > > 同时重试超过次数，任务失败，如下图： > > 我想问一下，flink重试机制中 >

Re: Flink DataStream 作业如何获取到作业血缘？

通过 JobGraph 可以获得 transformation 信息，可以获得具体的 Source 或者 Doris Sink，之后再通过反射获取里面的 properties 信息进行提取。 可以参考 OpenLineage[1] 的实现. 1. https://github.com/OpenLineage/OpenLineage/blob/main/integration/flink/shared/src/main/java/io/openlineage/flink/visitor/wrapper/FlinkKafkaConsumerWrapper.java Best,

Re: Flink Prometheus Connector问题

我理解可以参考 FLIP 中的设计， 基于 Prometheus Remote-Write API v1.0 来初步实现一个 SinkFunction 实现写入 Prometheus Best, Feng On Fri, Feb 23, 2024 at 7:36 PM 17610775726 <17610775...@163.com> wrote: > Hi > 参考官网， >

Re: flink sql中的自定义sink connector如何获取到source table中定义的event time和watermark?

我理解不应该通过 rowData 获取， 可以通过 Context 获得 watermark 和 eventTime. Best, Feng On Tue, Feb 20, 2024 at 4:35 PM casel.chen wrote: > 请问flink sql中的自定义sink connector如何获取到source table中定义的event time和watermark? > > > public class XxxSinkFunction extends RichSinkFunction implements > CheckpointedFunction,

Re: Flink任务链接信息审计获取

我理解应该是平台统一配置在 flink-conf.yaml 即可， 不需要用户单独配置相关参数. Best, Feng On Sun, Feb 4, 2024 at 11:19 AM 阿华田 wrote: > 看了一下 这样需要每个任务都配置listener，做不到系统级的控制，推动下游用户都去配置listener比较困难 > > > | | > 阿华田 > | > | > a15733178...@163.com > | > 签名由网易邮箱大师定制 > > > 在2024年02月2日 19:38，Feng Jin 写道： > hi, > > 可以参考下

Re: Flink任务链接信息审计获取

hi, 可以参考下 OpenLineage[1] 的实现， 通过 Flink 配置JobListener 拿到 Transformation 信息，然后解析 Source 和 Sink 拿到血缘信息。 [1] https://github.com/OpenLineage/OpenLineage/blob/main/integration/flink/src/main/java/io/openlineage/flink/OpenLineageFlinkJobListener.java Best, Feng On Fri, Feb 2, 2024 at 6:36 PM 阿华田

Re: flink ui 算子数据展示一直loading...

可以尝试着下面几种方式确认下原因： 1. 打开浏览器开发者模式，看是否因为请求某个接口卡住 2. 查看下 JobManager 的 GC 情况，是否频繁 FullGC 3. 查看下 JobManager 的日志，是否存在某些资源文件丢失或者磁盘异常情况导致 web UI 无法访问. Best, Feng On Tue, Jan 23, 2024 at 6:16 PM 阿华田 wrote: > > >

RE: Re:RE: RE: flink cdc动态加表不生效

Hi, oracle cdc connector 已经接入增量快照读框架，动态加表也是可以统一去实现的。 可以去社区创建issue，也欢迎直接贡献。 Best, Jiabao On 2024/01/19 04:46:21 "casel.chen" wrote: > > > > > > > 想知道oracle cdc connector不支持动态加表的原因是什么？可否自己扩展实现呢？ > > > > > > > > > > > > 在 2024-01-19 11:53:49，"Jiabao Sun" 写道： > >Hi, > > > >Oracle

Re:RE: RE: flink cdc动态加表不生效

想知道oracle cdc connector不支持动态加表的原因是什么？可否自己扩展实现呢？ 在 2024-01-19 11:53:49，"Jiabao Sun" 写道： >Hi, > >Oracle CDC connector[1] 目前是不支持动态加表的。 > >Best, >Jiabao > >[1] >https://ververica.github.io/flink-cdc-connectors/release-2.4/content/connectors/oracle-cdc.html > > >On 2024/01/19

RE: RE: flink cdc动态加表不生效

Hi, Oracle CDC connector[1] 目前是不支持动态加表的。 Best, Jiabao [1] https://ververica.github.io/flink-cdc-connectors/release-2.4/content/connectors/oracle-cdc.html On 2024/01/19 03:37:41 Jiabao Sun wrote: > Hi, > > 请提供一下 flink cdc 的版本，使用的什么连接器。 > 如果方便的话，也请提供一下日志。 > 另外，table 的正则表达式可以匹配到新增的表吗？ > >

RE: flink cdc动态加表不生效

Hi, 请提供一下 flink cdc 的版本，使用的什么连接器。 如果方便的话，也请提供一下日志。 另外，table 的正则表达式可以匹配到新增的表吗？ Best, Jiabao [1] https://ververica.github.io/flink-cdc-connectors/release-3.0/content/connectors/mysql-cdc%28ZH%29.html#id15 On 2024/01/19 03:27:22 王凯 wrote: > 在使用flink

Re: flink cdc 读取数据后面可以跟窗口函数吗

你好， CDC Source 目前不支持窗口函数。 不过可以考虑通过非窗口聚合的方式实现类似的效果。具体方法为： 1. 使用DATE_FORMAT函数，将时间字段转换成分钟粒度的字符串，作为窗口值。 2. 根据窗口值进行GROUP BY聚合。 Best, Hang Xuyang 于2024年1月17日周三 19:34写道： > Hi, > Flink SQL中可以用Group Window[1]的方式来读完cdc数据后加窗口。 > 可以具体描述一下“一直不生效”的现象和SQL么？ > > > > [1] >

Re: flink-checkpoint 问题

> > > > > 任务人为从25548恢复时失败，抛出异常找不到_metadate文件 > > > | | > 吴先生 > | > | > 15951914...@163.com > | > 回复的原邮件 > | 发件人 | Xuyang | > | 发送日期 | 2024年1月11日 14:55 | > | 收件人 | | > | 主题 | Re:回复： flink-checkpoint 问题 | > Hi, 你的图挂了，可以用图床处理一下，或者直接贴log。

Re: flink-checkpoint 问题

你好， 方便的话贴一下jobmanager的log吧，应该有一些线索 On Wed, Jan 10, 2024 at 5:55 PM 吴先生 <15951914...@163.com> wrote: > Flink版本： 1.12 > checkpoint配置：hdfs > > 现象：作业由于一些因素第N个checkpoint失败，导致任务重试，任务重试失败，hdfs中不存在第N个chk路径，但是为什么会出现一个第N+1的chk路径，且这个路径下是空的 > >

Re: Flink CDC中如何在Snapshot阶段读取数据时进行限流？

Hi, GuavaFlinkConnectorRateLimiter 目前只在 flink-connector-gcp-pubsub[1] 有使用。 Flink CDC 还未支持限流[2]，目前可以尝试降低 snapshot 并发数来缓解数据库压力。 Best, Jiabao [1]

RE: Flink SQL Windowing TVFs

Hi, 在 1.14.0 版本中，CUMULATE 函数是需要用在GROUP BY聚合场景下的[1]。 部署到生产的 SQL 是否包含了 GROUP BY 表达式？ 本地测试的Flink版本是不是1.14.0? Best, Jiabao [1] https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/table/sql/queries/window-tvf/#cumulate On 2023/12/29 04:57:09 "jiaot...@mail.jj.cn" wrote: > Hi，

Re: flink cdc 3.0 schema evolution原理是怎样的？

Hi, 是的，目前来说会 block 住。 flush + apply schema change 一般来说不会持续太长时间， 且 schema 变更一般来说是低频事件，即使 block 也不会有太大性能影响。 Best, Jiabao > 2023年12月28日 12:57，casel.chen 写道： > > > > > 感谢解惑！ > 还有一个问题：如果一个 pipeline 涉及多张表数据同步，而只有一个表出现 schema 变更的话，其他表的数据处理也会 block 住吗？ > > > > > > > > > 在 2023-12-28

Re: flink cdc 3.0 schema evolution原理是怎样的？

Hi, > 为什么最后output.collect(new StreamRecord<>(schemaChangeEvent)); > 还要发送一次SchemaChangeEvent呢？ Sink 也会收到 SchemaChangeEvent，因为 Sink 可能需要根据 Schema 变更的情况来调整 serializer 或 writer，参考 DorisEventSerializer > 最后一行requestReleaseUpstream()执行被block的原因是什么？是如何hold upstream然后再release > upstream的呢？ 被 block

Re: Re: Flink CDC MySqlSplitReader问题

Hi， 我记得这段逻辑是为了保证在新增表后，binlog 读取能和新增表的快照读取一起进行，保证binlog读取不会中断。 这里应该是会先读binlog，然后再读snapshot，再是binlog。这样的切换，来保证binlog 能一直有数据读出来。 Best， Hang casel.chen 于2023年12月22日周五 10:44写道： > 那意思是会优先处理已经在增量阶段的表，再处理新增快照阶段的表？顺序反过来的话会有什么影响？如果新增表全量数据比较多会导致其他表增量处理变慢是么？ > > > > > > > > > > > > > > > > > > 在

Re: Flink CDC MySqlSplitReader问题

Hi，casel 这段逻辑应该只有在处理到新增表的时候才会用到。 CDC 读取数据正常是会在所有SnapshotSplit读取完成后，才会下发BinlogSplit。 但是如果是在增量阶段重启作业，同时新增加了一些表，就会出现同时有BinlogSplit和SnapshotSplit的情况，此时才会走到这段逻辑。 Best， Hang key lou 于2023年12月20日周三 16:24写道： > 意思是当 有 binlog 就意味着 已经读完了 snapshot > > casel.chen 于2023年12月19日周二 16:45写道： > > >

Re: Flink CDC MySqlSplitReader问题

意思是当 有 binlog 就意味着 已经读完了 snapshot casel.chen 于2023年12月19日周二 16:45写道： > 我在阅读flink-connector-mysql-cdc项目源码过程中遇到一个不清楚的地方，还请大佬指点，谢谢！ > > > MySqlSplitReader类有一段代码如下，注释“(1) Reads binlog split firstly and then read > snapshot split”这一句话我不理解。 > 为什么要先读binlog split再读snapshot

Re: Flink-1.15版本

看起来是类似的，不过这个报错应该是作业失败之后的报错，看还有没有其他的异常日志。 Best, Feng On Sat, Nov 4, 2023 at 3:26 PM Ray wrote: > 各位专家：当前遇到如下问题1、场景：在使用Yarn场景下提交flink任务2、版本：Flink1.15.03、日志：查看yarn上的日志发下，版本上的问题2023-11-04 > 15:04:42,313 ERROR org.apache.flink.util.FatalExitExceptionHandler > [] - FATAL: Thread

Re: flink sql如何实现json字符数据解析？

Flink SQL比较适合处理结构化的数据，不知道你的body_data中的filed数量是否是固定的。如果是固定的，那可以将源和目标的格式写成Table形式。 比如： SourceT: ( uuid String, body_data ARRAY> ) SinkT ( result ARRAY> ) Insert into SinkT (result) select Array[ROW(uuid, null,body_data[1]. field1 as body_data.fild1, body_data[1]. Field2

Re: flink sql作业如何支持配置流？

Hi casel, 我们在生产中有类似的做法，可以考虑实现一个udtf，监听apollo的配置，根据配置选择是否filter数据。 Best, Yu Chen > 2023年11月20日 21:05，Xuyang 写道： > > Hi, >是否可以将这个”配置维表“换成流表，利用flink cdc，改动这个配置表的时候，监听字段cdc变化，同时下游上流join呢？ > > > > > -- > >Best！ >Xuyang > > > > > > 在 2023-11-20 19:24:47，"casel.chen" 写道： >>

Re: Re: flink 1.18.0 使用 hive beeline + jdbc driver连接sql gateway失败

hiveserver2 endpoint 就是让 flink gateway 直接变成 hive server2，对外来讲它就是 hive server2 了，它可以直接跟已有的跟 hive server2 的工具配合一起使用。 但是现在你其实用的是 flink jdbc driver，这个并不是跟 hive server2 交互，它就是跟 flink gateway 交互，所以你用hive server2的模式启动，它就不认识了。 casel.chen 于2023年10月30日周一 14:36写道： > > 果然不指定endpoint为hiveserver2类型后使用hive

Re: flink sql如何处理脏数据问题？

还有一种做法就是使用datastream，datastream支持sideoutput，但 flink sql不支持，不过有一种迂回的做法就是flinksql -> datastream -> flink sql，可以查一下官网资料，flinksql和datastream可以互相转换。 Xuyang 于2023年10月30日周一 10:17写道： > Flink SQL目前对于脏数据没有类似side output的机制来输出，这个需求用自定义connector应该可以实现。 > > > > > > > > -- > > Best！ > Xuyang > > > > >

Re: flink 1.18.0 使用 hive beeline + jdbc driver连接sql gateway失败

Hi casel, Flink JDBC 链接到 gateway 目前使用的是 flink 的 gateway 接口，所以你在启动 gateway 的时候不用指定 endpoint 为 hiveserver2 类型，用 Flink 默认的 gateway endpoint 类型即可。 casel.chen 于2023年10月29日周日 17:24写道： > > 1. 启动flink集群 > bin/start-cluster.sh > > > 2. 启动sql gateway > bin/sql-gateway.sh start

Re: flink sql不支持show create catalog 吗？

hi casel 从 1.18 开始，引入了 CatalogStore，持久化了 Catalog 的配置，确实可以支持 show create catalog 了。 Best, Feng On Fri, Oct 20, 2023 at 11:55 AM casel.chen wrote: > 之前在flink sql中创建过一个catalog，现在想查看当初创建catalog的语句复制并修改一下另存为一个新的catalog，发现flink > sql不支持show create catalog 。 > 而据我所知doris是支持show create

Re: Flink SQL的状态清理

:01 PM 小昌同学 wrote: > 你好，老师，我也是这样设置的，我这边是flink sql client，但是我去flink web ui界面并没有看到这个配置生效。 > > > | | > 小昌同学 > | > | > ccc0606fight...@163.com > | > 回复的原邮件 > | 发件人 | Jane Chan | > | 发送日期 | 2023年9月25日 11:24 | > | 收件人 | | > | 主题 | Re: Flink SQL的状态清理 |

Re: flink两阶段提交

hi, 可以参考这篇博客，描述的非常清晰： https://flink.apache.org/2018/02/28/an-overview-of-end-to-end-exactly-once-processing-in-apache-flink-with-apache-kafka-too/ Best, Feng On Sun, Sep 24, 2023 at 9:54 PM 海风 <18751805...@163.com> wrote: > 请教一下，flink的两阶段提交对于sink算子，预提交是在做检查点的哪个阶段触发的？预提交时具体是做了什么工作？ > > >

Re: Flink CDC消费Apache Paimon表

hi casel Flink 实时消费 paimon，默认情况就是全量 + 增量的方式。 具体可以参考： https://paimon.apache.org/docs/master/maintenance/configurations/ 中的 scan.mode 参数 best, Feng On Fri, Sep 29, 2023 at 5:50 PM casel.chen wrote: > 目前想要通过Flink全量+增量消费Apache Paimon表需要分别起离线和增量消费两个作业，比较麻烦，而且无法无缝衔接，能否通过类似Flink >

Re: Flink SQL的状态清理

Hi, 可以通过设置 table.exec.state.ttl 来控制状态算子的 state TTL. 更多信息请参阅 [1] [1] https://nightlies.apache.org/flink/flink-docs-master/zh/docs/dev/table/concepts/overview/#%e7%8a%b6%e6%80%81%e7%ae%a1%e7%90%86 Best, Jane On Thu, Sep 21, 2023 at 5:17 PM faronzz wrote: > 试试这个

Re: Flink cdc 2.0 历史数据太大，导致log积压怎么解决

你好，除了这些运维手段外，flink cdc本身有什么解法吗，比如说增量阶段不用从头开始读binlog，因为其实很多都是重复读到的数据 > 2023年9月20日 21:00，Jiabao Sun 写道： > > Hi, > 生产环境的binlog还是建议至少保留7天，可以提高故障恢复时间容忍度。 > 另外，可以尝试增加snapshot的并行度和资源来提升snapshot速度，snapshot完成后可以从savepoint恢复并减少资源。 > Best, > Jiabao >

Re: Flink cdc 2.0 历史数据太大，导致log积压怎么解决

Hi, 生产环境的binlog还是建议至少保留7天，可以提高故障恢复时间容忍度。 另外，可以尝试增加snapshot的并行度和资源来提升snapshot速度，snapshot完成后可以从savepoint恢复并减少资源。 Best, Jiabao -- From:jinzhuguang Send Time:2023年9月20日(星期三) 20:56 To:user-zh Subject:Flink cdc 2.0 历史数据太大，导致log积压怎么解决

Re: flink-metrics如何获取applicationid

退订 On Wed, Aug 30, 2023 at 19:14 allanqinjy wrote: > hi， >请教大家一个问题，就是在上报指标到prometheus时候，jobname会随机生成一个后缀，看源码也是new Abstract > ID（），有方法在这里获取本次上报的作业applicationid吗？

Re: flink rocksdb在托管&非托管模式下rocksdb内存计算

Hi, https://flink-learning.org.cn/article/detail/c1db8bc157c72069979e411cd99714fd 这篇文章中有一些关于 Flink RocksDB write buffer 和 block cache 内存计算的理论和实例讲解，可以参考下 On Fri, Sep 1, 2023 at 2:56 PM crazy <2463829...@qq.com.invalid> wrote: > 大佬们好， >flink1.13.5 >

Re: flink-metrics如何获取applicationid

hi, 可以尝试获取下 _APP_ID 这个 JVM 环境变量. System.getenv(YarnConfigKeys.ENV_APP_ID); https://github.com/apache/flink/blob/6c9bb3716a3a92f3b5326558c6238432c669556d/flink-yarn/src/main/java/org/apache/flink/yarn/YarnConfigKeys.java#L28 Best, Feng On Wed, Aug 30, 2023 at 7:14 PM allanqinjy wrote: > hi，

Re: flink sql语句转成底层处理函数

-zh@flink.apache.org | > | 抄送至 | | > | 主题 | Re: flink sql语句转成底层处理函数 | > 如果想看一个sql被转换后包含哪些具体执行步骤，可以通过explain语法[1]查看执行计划 > > [1] > > https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/explain/ > > On Sun, Aug 27, 2023 at 5:23 PM 海风 <18751805...@163.com

Re: flink sql语句转成底层处理函数

如果想看一个sql被转换后包含哪些具体执行步骤，可以通过explain语法[1]查看执行计划 [1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/explain/ On Sun, Aug 27, 2023 at 5:23 PM 海风 <18751805...@163.com> wrote: > 请教下，是否可以去查询一个flink > sql提交运行后，flink给它转成的底层处理函数到底是什么样的，假如涉及状态计算，flink给这个sql定义的状态变量是哪些呢？ > > >

Re: flink sql作业状态跨存储系统迁移问题

可以 savepoint 到 HDFS，然后配置 checkpoint 的地址为 对象存储。 我们就是 flink 支持对象存储和 HDFS。 Hangxiang Yu 于2023年8月2日周三 14:03写道： > Hi, 我理解可以有两种方式： > 1. 设定从某个存储集群上恢复并向另一个存储集群上快照，即设置[1]为 HDFS地址，[2] 为后面的对象存储地址 > 2. 还是在HDFS集群上启停作业，设置 savepoint 目录[3]到对象存储 > > 关于 state processor api，目前 sql 作业确实操作起来比较困难，只能从日志里获取 uid

Re: Flink 窗口触发条件

hi, 感觉和[1]的问题比较像，事件时间的window在onElement和onEventTime时会触发，这两个方法又会根据watermark判断，可以看看o.a.f.table.runtime.operators.window.triggers包和o.a.f.table.runtime.operators.wmassigners包。 [1] https://juejin.cn/post/6850418110010179597 小昌同学 于2023年8月10日周四 10:52写道： > >

Re: Flink消费MySQL

; > | > > ccc0606fight...@163.com > > | > > 回复的原邮件 > > | 发件人 | Shammon FY | > > | 发送日期 | 2023年8月8日 10:37 | > > | 收件人 | | > > | 主题 | Re: Flink消费MySQL | > > Hi， > > > > 你代码里的ResultSet读取完成后需要将resultSet关闭掉，避免资源泄漏 > >

Re: Flink消费MySQL

; > > | | > 小昌同学 > | > | > ccc0606fight...@163.com > | > 回复的原邮件 > | 发件人 | Shammon FY | > | 发送日期 | 2023年8月8日 10:37 | > | 收件人 | | > | 主题 | Re: Flink消费MySQL | > Hi， > > 你代码里的ResultSet读取完成后需要将resultSet关闭掉，避免资源泄漏 > > 至于你提到的Mysql数据读完程序就结束

Re: Flink消费MySQL

Hi， 你代码里的ResultSet读取完成后需要将resultSet关闭掉，避免资源泄漏 至于你提到的Mysql数据读完程序就结束具体是指哪块？mysql是bounded source，数据消费完成并且整个作业计算完成后，就会结束，这是正常情况 Best, Shammon FY On Mon, Aug 7, 2023 at 5:04 PM 小昌同学 wrote: > 各位老师好 > ，我这边在本地使用通过继承RichSourceFunction类，实现从MySQL中读取数据，但是为啥程序将MySQL中的全部数据获取出来后，程序就自动停止了啊； > 以下是我的代码： > | >

Re: flink作业如何从yarn平滑迁移到k8s？

你好， 如果你们也使用的是官方的Flink Kubernetes Operator，可以参考我们迁移的经验：迁移的时候设置FlinkDeployment的initalSavepoint为HDFS上Savepoint的地址，同时配置savepoint/checkpoint目录为OSS。这样Flink启动的时候会从HDFS中的状态恢复，并将新的checkpoint保存在oss中。 On Sun, Aug 6, 2023 at 10:03 PM casel.chen wrote: > flink on

Re: Flink窗口状态清除疑问

首先你窗口是30min，刚刚开始肯定会是涨的。 其次，后续稳定后，继续涨可能是因为流量在变化。 最后，流量不变情况下，还可能受到延迟的影响。 lxk 于2023年7月25日周二 11:22写道： > > 相关配置： > Flink:1.16 > > | Checkpointing Mode | Exactly Once | > | Checkpoint Storage | FileSystemCheckpointStorage | > | State Backend | EmbeddedRocksDBStateBackend | > | Interval | 8m 0s | > > >

Re: Flink ML

这个取决于你是什么模型，比如python中sklearn的大多模型都可以导出成pmml格式模型，然后java用jpmml库就可以导入进行预测。 如果是tensorflow模型，也有，只不过我忘记了，你可以找找。 15904502343 <15904502...@163.com> 于2023年8月1日周二 16:48写道： > > 您好 > 我想知道是否有代码示例，可以在Flink程序中加载预先训练好的编码模型(用python编写)

Re: flink sql作业状态跨存储系统迁移问题

Hi, 我理解可以有两种方式： 1. 设定从某个存储集群上恢复并向另一个存储集群上快照，即设置[1]为 HDFS地址，[2] 为后面的对象存储地址 2. 还是在HDFS集群上启停作业，设置 savepoint 目录[3]到对象存储 关于 state processor api，目前 sql 作业确实操作起来比较困难，只能从日志里获取 uid 等信息，以及理解 sql 实际产生的状态才能使用； [1]

Re: flink-job-history 任务太多页面卡死

Hi Flink UI 需要加载所有的 Job 信息并在 UI 渲染，在作业比较多的时候很容易导致 UI 卡死。 不只在这个页面，在一些并发比较大的任务上打开 subtask 页面也很容易导致UI 卡死。 Flink UI 需要一个分页的功能来减少数据加载和 UI 渲染的压力 Best, Weihua On Fri, Jul 28, 2023 at 11:29 AM Shammon FY wrote: > Hi, > > > 可以通过配置`jobstore.max-capacity`和`jobstore.expiration-time`控制保存的任务数，具体参数可以参考[1]

Re: flink-job-history 任务太多页面卡死

Hi, 可以通过配置`jobstore.max-capacity`和`jobstore.expiration-time`控制保存的任务数，具体参数可以参考[1] [1] https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#full-jobmanager-options Best, Shammon FY On Fri, Jul 28, 2023 at 10:17 AM 阿华田 wrote: > 目前flink-job-history >

RE: flink如何正确使用mybatis

SqlSession 需要关闭，建议使用 SqlSessionManager，可以不用手动关闭 SqlSession。 On 2023/07/18 02:13:16 lxk wrote: > 在flink内需要使用mybatis做些简化查询的工作，目前我的使用方式如下 > > public class MybatisUtil { > > private static final Logger LOGGER = > LogFactory.createNewLogger("MybatisUtil"); > private static ThreadLocal tl = new

Re: Flink connector 是否支持忽略delete message

Hi, shi franke. 你可以尝试自己实现一个 DynamicTableSink，在里面添加参数 “sink.ignore-delete”。 你可以参考 github 上的一些实现，例如 clickhouse： https://github.com/liekkassmile/flink-connector-clickhouse-1.13 shi franke 于2023年7月7日周五 19:24写道： > >

Re: Flink connector 是否支持忽略delete message

感谢您的回复，这样自定义是可以实现的，我们目前使用的是1.15的flink版本。想看一下社区是不是有在框架层面实现这个配置的支持，理解这应该也是一个相对common的配置 junjie.m...@goupwith.com 于2023年7月7日周五 17:57写道： > 可以自己用DataStream API通过RowKind进行过滤。 > 如下示例代码：import org.apache.flink.api.common.functions.RichFlatMapFunction; > import org.apache.flink.types.Row; > import

Re: Flink 1.16 流表 join 的 FilterPushDown 及并行

Hi, Chai Kelun, 你的 filter condition 里面包含了你自定义的 UDF，是不满足 filter push down 的条件的，因为对于优化器来说 UDF 是不确定的，优化器不能从里面提取到可以下推的条件, 如果你想实现下推，可以尝试抽取下确定性的 condition，如 product.id > 10 etc.。另外，Flink 是支持 broadcast hash join 的，如果你想控制某两个表的 join type，你可以通过 join hint 来指定 join 类型为 broadcast。() Chai Kelun 于2023年7月3日周一

Re: flink on native k8s里如何使用flink sql gateway

Hi, 我们的做法是启动Flink集群后，在其他节点(pod或者独立启动)启动Sql-Gateway，通过Flink的地址远程连接Flink集群，这样Sql-Gateway的部署和Flink集群完全分开 Best, Shammon FY On Tue, Jul 4, 2023 at 10:52 AM chaojianok wrote: > 大家好，请教个问题。 > > 用native kubernetes方式在k8s集群上部署好了flink，现在需要在这个flink集群里使用flink sql > gateway，大家有什么好的方案吗？ > 目前的做法是，进入pod里启动sql

Re: flink sql作业指标名称过长把prometheus内存打爆问题

退订 On Thu, Jun 15, 2023 at 7:23 PM im huzi wrote: > 退订 > > On Tue, Jun 13, 2023 at 08:51 casel.chen wrote: > > > 线上跑了200多个flink > > > sql作业，接了prometheus指标（prometheus定期来获取作业指标）监控后没跑一会儿就将prometheus内存打爆（开了64GB内存），查了一下是因为指标名称过长导致的。 > > flink > > >

Re: flink sql作业指标名称过长把prometheus内存打爆问题

退订 On Tue, Jun 13, 2023 at 08:51 casel.chen wrote: > 线上跑了200多个flink > sql作业，接了prometheus指标（prometheus定期来获取作业指标）监控后没跑一会儿就将prometheus内存打爆（开了64GB内存），查了一下是因为指标名称过长导致的。 > flink > sql作业的指标名称一般是作业名称+算子名称组成的，而算子名称是由sql内容拼出来的，在select字段比较多或sql较复杂的情况下容易生成过长的名称， > 请问这个问题有什么好的办法解决吗？

Re: flink写kafka 事务问题

消费者要设置事务隔离级别 > 2023年6月15日 16:23，163 写道： > > 据我了解，kafka支持事务，开启checkpoint及exactly-once后仅当checkpoint执行完毕后才能将数据写入kafka中。测试：flink读取kafka的topic > a写入topic b，开启checkpoint及exactly-once，flink未执行完新一次的checkpoint，但topic > b已经可以消费到新数据，这是什么原因？请大家指教！

Re: Re: flink sql作业指标名称过长把prometheus内存打爆问题

配置参数之后， task name 也会简化. Best, Feng On Wed, Jun 14, 2023 at 11:23 AM casel.chen wrote: > > > > > > > > > > > > > 谢谢，除了operator name，我看了flink sql作业生成的task name也很长，目前有办法可以简化下吗？例如 > > >

Re: flink sql作业指标名称过长把prometheus内存打爆问题

hi casel 1. 可以考虑使用 Flink1.15, 使用精简的 operator name https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/table/config/#table-exec-simplify-operator-name-enabled 2. Flink 也提供了 restful 接口直接获取瞬时的 metric，如果不需要历史的 metric

Re: flink on yarn rocksdb内存超用

Hi, 目前对RocksDB使用的内存是没有严格限制住的，可以参考这个 ticket： https://issues.apache.org/jira/browse/FLINK-15532 如果要定位到内存使用情况，可以先看一些粗的Metrics： https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/config/#rocksdb-native-metrics 如果要再细致定位到单 instance 内部 RocksDB 的详细内存使用情况，可能需要用 malloc

Re: flink 输出异常数据

; > > > > > | | > > 小昌同学 > > | > > | > > ccc0606fight...@163.com > > | > > 回复的原邮件 > > | 发件人 | Weihua Hu | > > | 发送日期 | 2023年5月29日 15:29 | > > | 收件人 | | > > | 主题 | Re: flink 输出异常数据 | > > Hi, > > > > 你使用的数据源是什么呢？Kafka

Re: flink 输出异常数据

这个得靠你自己打日志吧，在可能出NPE的地方 try catch 到，然后打印原始记录。 小昌同学 于2023年5月29日周一 18:30写道： > > 你好，数据源是kafka，使用的是stream api > > > | | > 小昌同学 > | > | > ccc0606fight...@163.com > | > 回复的原邮件 > | 发件人 | Weihua Hu | > | 发送日期 | 2023年5月29日 15:29 | > | 收件人 | | > |

Re: flink jdbcsink 连接数的问题

你好，这个问题和flink无关，看你主键实现机制吧，如果是自增，那就是mysql级别自动实现的自增，跟flink搭不上关系的。 小昌同学 于2023年5月31日周三 09:41写道： > > 老师，你好，再请教一下，连接数与并行度有关系的话，如果插入数据的MySQL是有主键的话，是不是连接数据也就是并行度只能为1啦呀，如果是多个并行度的话，可能会造成主键冲突； > 感谢各位老师的指导 > > > | | > 小昌同学 > | > | > ccc0606fight...@163.com > | > 回复的原邮件 > | 发件人 | lxk | > | 发送日期 |

Re: flink web ui显示问题

> | 发件人 | Shammon FY | > | 发送日期 | 2023年5月31日 09:59 | > | 收件人 | | > | 主题 | Re: flink web ui显示问题 | > Hi, > > 好像没有收到附件或者文档，你可以检查确认一下 > > Best, > Shammon FY > > On Wed, May 31, 2023 at 9:52 AM 小昌同学 wrote: > > 各位老师好，请教一个关于flink web ui的显示问题； > 具体的显示异常截图的我

Re: flink web ui显示问题

Hi, 好像没有收到附件或者文档，你可以检查确认一下 Best, Shammon FY On Wed, May 31, 2023 at 9:52 AM 小昌同学 wrote: > 各位老师好，请教一个关于flink web ui的显示问题； > 具体的显示异常截图的我以附件的形式放在文档中，我的疑惑是web > ui上面已经显示watermark，但是看detail的时候显示不是watermark； > 感谢各位老师指导 > > 小昌同学 > ccc0606fight...@163.com > >

Re: flink 输出异常数据

Hi, 你使用的数据源是什么呢？Kafka 吗？用的是 FlinkSQL 还是 DataStream API 呢？ 方便把异常栈贴一下吗 Best, Weihua On Mon, May 29, 2023 at 1:36 PM 小昌同学 wrote: > > 各位老师，我有一个作业运行很久了，但是最近源系统有一些脏数据导致作业运行失败，看yarn的日志报错是空指针，但是我现在想把那一条脏数据捕获到，请问一下有啥办法吗？谢谢各位老师的指导 > > > | | > 小昌同学 > | > | > ccc0606fight...@163.com > |

Re: Flink RocketMQ Connector

hi casel Flink RocketMQ connector 是由 RockeMQ 社区维护的， 对应的项目地址是: https://github.com/apache/rocketmq-flink 这个版本默认的消息是格式 DELIMIT 格式(默认消息是 String，按分隔符进行分割)， 只能指定消息的列分隔符. best, feng On Fri, May 26, 2023 at 7:44 PM casel.chen wrote: > 有没有Flink RocketMQ官方连接器?

Re: flink 窗口触发计算的条件

如果你只发送了一条数据，那么watermark不会推进，就不会触发窗口计算。你需要更多数据。 小昌同学 于2023年5月25日周四 09:32写道： > > 各位老师，请教一下关于flink 事件时间窗口的执行时间点的相关问题； > 我使用的窗口是：TumblingEventTimeWindows(Time.minutes(1L)),我使用的时间定义是System.currentTimeMillis()，watermark是2秒， > 但是当我发送一条数据后，过了5分钟之后，窗口都没有触发计算，想请各位老师帮忙看一下程序的问题所在： > 相关代码以及样例数据如下： > | >

Re: Flink提交作业是否可以跳过上传作业jar包这一步？

可以考虑基于flink-kubernetes依赖下的KubernetesClusterDescriptor来启动任务，可以参考https://github.com/collabH/flink-deployer/blob/main/infrastructure/src/main/java/com/flink/plugins/inf/deployer/KubernetesClusterDeployer.java > 2023年5月15日 19:21，casel.chen 写道： > > 我们开发了一个实时计算平台提交flink >

Re:Re: Re: Flink广播流状态清理策略不生效

好的，感谢 在 2023-05-15 15:49:12，"Hangxiang Yu" 写道： >Hi, 可以参考这个 Ticket ，就是讨论要给 Broadcast State 加 TTL 的，当时应该没有继续深入讨论： >https://issues.apache.org/jira/browse/FLINK-13721 >方便的话你可以在 Ticket 下面也分享下你的使用场景、观察到的现象吗？也可以在 Ticket 下 Vote for this issue. >我这边也会帮忙一起看下 > >On Mon, May 15, 2023 at 1:41 

Re: Re: Flink广播流状态清理策略不生效

Hi, 可以参考这个 Ticket ，就是讨论要给 Broadcast State 加 TTL 的，当时应该没有继续深入讨论： https://issues.apache.org/jira/browse/FLINK-13721 方便的话你可以在 Ticket 下面也分享下你的使用场景、观察到的现象吗？也可以在 Ticket 下 Vote for this issue. 我这边也会帮忙一起看下 On Mon, May 15, 2023 at 1:41 PM lxk wrote: > 这么看来，广播流好像不适合在生产中使用，状态会无限止的增长。这块官方有计划增加ttl功能吗。 >

Re: Flink广播流状态清理策略不生效

Hi, 目前像 Broadcast state 这种 Operator State 应该是不支持 TTL 设置的，可以参考这里 对 State TTL 的描述； On Mon, May 15, 2023 at 11:05 AM lxk wrote: > flink版本:1.14 >

Re: flink 状态设置

Hi, "如果不对于状态进行管理，后续程序会出现问题"是指状态会变得太大？如果是这样，可以在group by的字段里增加一个天级的时间戳，这样就不会由于key被更新导致的状态过期失效问题 Best, Shammon FY On Fri, May 12, 2023 at 1:59 PM 小昌同学 wrote: > 各位老师好，我这边使用的flink sql是" > select funcId,funcIdDesc,serverIp,cast(min(maxTime-minTime) as > varchar(200)) as minTime,pk from > ( >

Re: flink 1.13 partition.time-extractor.timestamp-pattern 格式

Hi, 就像上面文档描述的，如果是多个字段组合成partition，可以在DDL中通过partition.time- extractor.timestamp-pattern将多个字段按照自己的partition格式需求进行组装。 CREATE TABLE fs_table ( user_id STRING, order_amount DOUBLE, dt STRING, `hour` STRING ) PARTITIONED BY (dt, `hour`) WITH ( 'connector'='filesystem', 'path'='...',

Re: flink issue可以登录，但是flink中文邮箱账号密码错误，是出现什么原因了嘛

> > flink issue可以登录 这个是jira账号吗？ flink中文邮箱账号密码 什么是flink中文邮箱账号 ?有无登陆页面链接 On Wed, Apr 19, 2023 at 11:36 AM kcz <573693...@qq.com.invalid> wrote: > 请帮忙看看是我哪里出问题了嘛？我的账号是kcz。我想咨询大佬flink avro的问题 > > > > > kcz > 573693...@qq.com > > > >

Re: Flink 误报checkpoint失败

hi, 扩缩容会重启作业，在作业重启期间，job manager 先启动了，还有部分task manager没启动就有可能报“Not all required tasks are currently running..”的错误，作业的所有task完全启动后这个错误就会消失。 Best, Yanfei Chen Yang 于2023年5月4日周四 09:44写道： > > 您好， > > 我的 Flink job是以 reactive 模式运行，然后用了 Kubernetes HPA 来自动扩容/缩容 > TaskManager。每当TaskManager >

Re: Flink rocksDB疑似内存泄露，导致被Linux kernel killed

Hi, 请问作业有配置ttl吗？ 另外可以参考下是否与下面两个问题类似： 1. pin L0 index in memory : https://issues.apache.org/jira/browse/FLINK-31089 2. max open files：https://issues.apache.org/jira/browse/FLINK-31225 Biao Geng 于2023年4月23日周日 15:35写道： > > Hi, > 可以配置下jemalloc来进行堆外内存泄漏的定位。 > 具体操作可以参考下这两篇文章。 >

Re: flink rocksdb异常

Hi 这是TM向JM发送消息超时了，可以了看下JM是否有错误日志，或者对应的TM和JM是否有资源打满等情况，导致akka消息超时 Best, Shammon FY On Sun, Apr 23, 2023 at 2:28 PM crazy <2463829...@qq.com.invalid> wrote: > Hi, 大佬好, >有个Flink on > Yarn程序，Flink版本使用的是flink-1.13.5，statebackend使用的是rocksdb，任务跑一段时间，就会出现如下堆栈异常： > > > 2023-04-20 22:32:08,127

  1   2   3   4   5   6   7   8   9   10   >