Re:Re: 提交任务不能指定第三方jar

Hi， 试试 -C,--classpath，我都是用这个提交UDF的 另外邮件列表发图片是看不到了，别再发截图了 Thanks， 在 2022-12-13 18:13:47，"melin li" 写道： 类似: spark-submit 支持--jars，更灵活方便， melin li 于2022年12月8日周四 11:09写道： 如果是作业依赖的jar，是可以打一个flat jar。有两种场景： 1、sql作业中，用户依赖某个connector jar，但平台不提供这个connector，需要用户上传， 2、自定义udf

Re:flink sql connector options如何支持Map数据类型？

Hi， > 遇到用户添加自定义请求头Headers问题 如果自定义Headers是和内容相关，那么就和connector options无关了，应该是需要用表的字段来描述 如果自定义Headers是定义table的时候确定的，那就是定义connector options的问题了 > 如何在connector options中支持Map数据类型呢？ options里面都是k=v的结构，v都是字符串，所以你要考虑的是如何用 字符串 描述 Map Thanks 在 2022-12-17 10:20:29，"casel.chen" 写道： >我想开发一个Flink SQL

Re:flink如何调用py模型

Hi, 我考虑过的几种方式： 1. 做成http服务，和flink分离，py模型本来跑起来也不快，做成http可以动态扩容，无状态的话 2. 用pyflink来跑任务，可以嵌python代码，就是任务启动非常慢，要复制虚拟环境，模型可以写成pandas的输入输出，这样模型也是可以独立开发的 3. Java调python的udf，py必须要能封装成函数，写udf毕竟麻烦 Thanks 在 2022-12-19 16:51:33，"kcz" <573693...@qq.com.INVALID> 写道： >flink调用py模型时候，大是采取什么方案，直接跟flink集成嘛？

为何雪花算法的udf 每次执行的结果都一样，如何能像UUID一样每次变化

环境信息 flink-1.13.6_scala_2.11 java 1.8 hive 1.1.0-cdh5.16.2 hbase 1.2.0-cdh5.16.2 使用的是yarn-session模式 UDF 类在附件中 应用场景是将hive表和kafka表join后输出到hbase，因为在每3秒倒序查询一次hbase的10条结果，故想将雪花算法生成的ID作为hbase的rowkey好做排序。 但是发现udf注册到flink之后，多次执行得到的结果是一样的，没有达到自动增长的目的，在sql-clinet下执行下面的SQL CREATE FUNCTION IF NOT EXISTS

如何查看Flink on Native K8s模式下失败作业的日志

Flink On Native K8s 模式下，如果流作业因异常失败了，作业的JobManager和TaskManager所在Pod都会消失，就无法查看作业日志。 请问在K8s模式下，在查看日志方面有没有相关解决方案。 目前我只想到将作业Jm和Tm打印的日志通过pv-pvc方式挂载NFS做持久化。这样做日志就可以保存下来方便查看。 -- Best， Hjw

退订

退订 | | 刘朝兵 |

[ANNOUNCE] Apache flink-connector-rabbitmq 3.0.0 released

The Apache Flink community is very happy to announce the release of Apache flink-connector-rabbitmq 3.0.0 This release marks the first time we have released this connector separately from the main Flink release. Over time more connectors will be migrated to this release model. This release is

[ANNOUNCE] Apache flink-connector-jdbc 3.0.0 released

The Apache Flink community is very happy to announce the release of Apache flink-connector-jdbc 3.0.0 This release marks the first time we have released this connector separately from the main Flink release. Over time more connectors will be migrated to this release model. This release is

[ANNOUNCE] Apache flink-connector-pulsar 3.0.0 released

The Apache Flink community is very happy to announce the release of Apache flink-connector-pulsar 3.0.0 This release marks the first time we have released this connector separately from the main Flink release. Over time more connectors will be migrated to this release model. This release is

来自Lynn Chen的邮件

退订

RE: Re: 【PyFlink】对于数据以Csv()格式写入kafka报错，以及使用python udf时无法启动udf

可以排查下是不是Flink_Lib_Home里自己“多手“引入了flink-python-x.x.jar，如果是的话，去掉就行，不然会导致beam包冲突 On 2020/11/03 01:44:00 jing wrote: > Hi, jincheng. > > 目前也遇到了类似问题，请问有什么思路吗？ > > flink-python 的 jar 都是有的，且版本是对的。 > > 版本是 1.11.1，这个主要是在引入了 udf 时出现的，之前是正常的。 > > 具体报错如下： > > Caused by: java.lang.NoClassDefFoundError: Could

退订

退订

Re:Re: flink sql connector options如何支持Map数据类型？

看过了，不支持http source table，而且即使http lookup table也不支持map数据类型 在 2022-12-19 14:51:42，"Weihua Hu" 写道： >Hi, 你可以尝试使用独立开源的 http connector > >https://github.com/getindata/flink-http-connector > >Best, >Weihua > > >On Sat, Dec 17, 2022 at 10:21 AM casel.chen wrote: > >> 我想开发一个Flink SQL

RE: Re:Re: sql-client pyexec参数生效疑问

设置-pyclientexec参数和sql client终端SET 'python.client.executable'='’; 都能解决` java.lang.IllegalStateException: Instantiating python function xx failed`的问题，但我接下来正常执行udf函数的时候依然报错，一直没有找到` Could not initialize class org.apache.beam.sdk.options.PipelineOptionsFactory `的原因是啥。 ``` select

关于flink sql调用python udf失败的问题

Flink使用官方docker-compose起的，python env用conda打的包（python3.7.12+apache-flink==1.16.0+apache-beam=2.38.0）； ```bash sql client 启动参数 # sql_client/task_manager/job_manager都挂载了对应目录，权限、所有者均为flink:flink bin/sql-client.sh \ --pyExecutable /opt/flink_data/requirements/py_env/jm_env/bin/python3.7 \ -pyfs

Re: flink sql connector options如何支持Map数据类型？

Hi, 你可以尝试使用独立开源的 http connector https://github.com/getindata/flink-http-connector Best, Weihua On Sat, Dec 17, 2022 at 10:21 AM casel.chen wrote: > 我想开发一个Flink SQL Http Connector，遇到用户添加自定义请求头Headers问题，如何在connector > options中支持Map数据类型呢？

退订

退订 | | 爱吃鱼 | | aichiyuyiy...@163.com |

flink sql connector options如何支持Map数据类型？

我想开发一个Flink SQL Http Connector，遇到用户添加自定义请求头Headers问题，如何在connector options中支持Map数据类型呢？

Re: [ANNOUNCE] Apache Flink Kubernetes Operator 1.3.0 released

Congratulations! Thanks a lot for the awesome work! Best regards, Biao Geng Őrhidi Mátyás 于2022年12月15日周四 01:30写道： > The Apache Flink community is very happy to announce the release of Apache > Flink Kubernetes Operator 1.3.0. > > Release highlights: > >- Upgrade to Fabric8 6.x.x and JOSDK

[ANNOUNCE] Apache Flink Kubernetes Operator 1.3.0 released

The Apache Flink community is very happy to announce the release of Apache Flink Kubernetes Operator 1.3.0. Release highlights: - Upgrade to Fabric8 6.x.x and JOSDK 4.x.x - Restart unhealthy Flink clusters - Contribute the Flink Kubernetes Operator to OperatorHub - Publish

Re: HiveServer2 Endpoint只支持Hive3.1以上版本吗

Hive2.3 也支持 Best regards, Yuxia - 原始邮件 - 发件人: "junjie miao" 收件人: "user-zh" 发送时间: 星期二, 2022年 12 月 13日 下午 6:23:15 主题: HiveServer2 Endpoint只支持Hive3.1以上版本吗 flink1.16.0中SQL Gateway HiveServer2 Endpoint只支持Hive3.1以上版本吗？有测试过其他hive版本吗？

Re: 提交任务不能指定第三方jar

类似: spark-submit 支持--jars，更灵活方便， [image: image.png] melin li 于2022年12月8日周四 11:09写道： > 如果是作业依赖的jar，是可以打一个flat jar。有两种场景： > 1、sql作业中，用户依赖某个connector jar，但平台不提供这个connector，需要用户上传， > 2、自定义udf 管理，依赖的jar 需要和任务一起提交。 > > yuxia 于2022年12月8日周四 10:06写道： > >> 为啥说 不能提交依赖的第三方jar？用户的 job 把这些包打进去不就好了吗？ 还是说你指的是

回复： flink-gelly官方文档

好的，谢谢 | | Jason_H | | hyb_he...@163.com | 回复的原邮件 | 发件人 | guozhi mang | | 发送日期 | 2022年12月13日 17:45 | | 收件人 | | | 主题 | Re: flink-gelly官方文档 | https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/libs/gelly/overview/ Jason_H 于2022年12月13日周二 17:39写道：

Re: flink-gelly官方文档

https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/libs/gelly/overview/ Jason_H 于2022年12月13日周二 17:39写道： > 请问一下，flink官网对应flink-gelly模块在哪里，请教大佬给个指引，半天找不到，网上搜到的是很老的版本。 > > > | | > Jason_H > | > | > hyb_he...@163.com > | -- Best regards

flink-gelly官方文档

请问一下，flink官网对应flink-gelly模块在哪里，请教大佬给个指引，半天找不到，网上搜到的是很老的版本。 | | Jason_H | | hyb_he...@163.com |

Flink 中的类加载KafkaSerializerWrapper

在使用kafka自定义序列化时会导致对应的class无法加载的问题。通过分析，现有代码在使用AppClassLoader类加载器先加载了KafkaSerializerWrapper时，用户提交任务自己编写的类是无法通过AppClassLoader加载的，通过FlinkUserCodeClassLoader加载的话需要KafkaSerializerWrapper也是通过FlinkUserCodeClassLoader加载。 public void open(InitializationContext context) throws Exception { final ClassLoader

Re: 使用flink 动态广播mysql数据报错

这没啥问题，但是你代码不能这么写，try catch放 while 内部去。放外边catch到异常不就退出循环了！ bigdata <1194803...@qq.com.invalid> 于2022年12月11日周日 15:12写道： > > java.lang.InterruptedException: sleep interrupted > at java.lang.Thread.sleep(Native Method) > at >

Re: 关于Flink重启策略疑惑

你好 1.Flink中(JM)JobMaster会监控各个Task的状态，如果Task由于某些原因失败了，JM触发failover,并且决策哪些task应该被重新启动。当然，如果JM挂掉的话，Flink支持配置高可用(HA),通过持久化一些信息到外部系统，从而做到通过standby JM正确接管作业。 2.无论单个Task挂掉还是TaskManager挂掉failover流程都可以正确处理，处理流程基本是一致的，TaskManager挂掉可以认为是上面所有被调度上去的Task fail了。 Best regards, Weijie 李义 于2022年12月9日周五

关于Flink重启策略疑惑

你好，我们团队在调研Flink相关技术。关于故障重启策略有些困惑 Task 故障恢复 | Apache Flink 1.故障重启是通过什么技术手段触发的，我搜查了很多资料 ，都仅提到重启策略是怎么配置的，但是谁触发的? 它不可能挂掉了自己重启吧？ 2.故障重启是Task级别还是作用于TaskManager服务？ 感谢并支持Flink开发者们的工作，Thanks!

flink kubernetes operator生产环境配置

hello all, flink kubernetes operator在生产环境的配置应该设置多少比较合适呢？目前我直接用helm去部署的，但是看到对应的pod中的两个container是没有设置request和limit cpu和memory和JVM内存的，我应该设置吗？规模在1000左右的任务应该设置多少会比较合适呢？ best wishes, tanjialiang. | | 谭家良 | | tanjl_w...@126.com |

Re: 1.15.2作业频繁（每 几十分钟 ~ 1小时）报 LocalTransportException: readAddress(..) failed: Connection timed out .

目前感觉和 https://issues.apache.org/jira/browse/FLINK-19249 和 https://issues.apache.org/jira/browse/FLINK-16030 有点类似。网络环境不稳定。相同配置在物理机没问题。 yidan zhao 于2022年12月7日周三 16:11写道： > > 谢谢，不过这几个参数和netty关系不大吧。 > heartbeat和akka的可能会和rpc超时有关，不过我这个是netty的报错，不是rpc部分。 > web和rest应该是和client提交任务有关。 > > Stan1005

Re: 提交任务不能指定第三方jar

如果是作业依赖的jar，是可以打一个flat jar。有两种场景： 1、sql作业中，用户依赖某个connector jar，但平台不提供这个connector，需要用户上传， 2、自定义udf 管理，依赖的jar 需要和任务一起提交。 yuxia 于2022年12月8日周四 10:06写道： > 为啥说 不能提交依赖的第三方jar？用户的 job 把这些包打进去不就好了吗？ 还是说你指的是 sql 作业？ > > Best regards, > Yuxia > > > 发件人: "melin li" > 收件人: "user-zh" > 发送时间: 星期四, 2022年 12

Re: 提交任务不能指定第三方jar

为啥说 不能提交依赖的第三方jar？用户的 job 把这些包打进去不就好了吗？ 还是说你指的是 sql 作业？ Best regards, Yuxia 发件人: "melin li" 收件人: "user-zh" 发送时间: 星期四, 2022年 12 月 08日 上午 9:46:45 主题: 提交任务不能指定第三方jar 客户端提交flink job 不能提交依赖的第三方jar，例如自定的函数jar，sql 里面的依赖connector jar，需要提前放置好。如果基于flink 平台化，需要动态的添加jar。目前可能的做法，就是把依赖的jar,

提交任务不能指定第三方jar

客户端提交flink job 不能提交依赖的第三方jar，例如自定的函数jar，sql 里面的依赖connector jar，需要提前放置好。如果基于flink 平台化，需要动态的添加jar。目前可能的做法，就是把依赖的jar, 动态的添加作业jar 的lib目录下。getJobJarAndDependencies 就是从jar 中获取依赖的jar。不是很方便。 是可以添加一个参数，指定依赖的jar，flink 设计各种诡异。 [image: image.png]

Re: flink 1.16 lookup join重试策略问题

如果结束时还未关联上，就视为当前记录不存在，按 inner join 过滤或 left join 补 null 值 https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/dev/table/sql/queries/hints/#%e5%bc%80%e5%90%af%e7%bc%93%e5%ad%98%e5%af%b9%e9%87%8d%e8%af%95%e7%9a%84%e5%bd%b1%e5%93%8d Best, Lincoln Lee casel.chen 于2022年12月7日周三

flink 1.16 lookup join重试策略问题

我们有场景会发生维表数据后于事实流表数据到达，使用flink 1.16 lookup join重试策略时，如果超过重试次数还没关联上会发生什么？待关联字段值都为null么？

Re:Re: 如何扩展flink sql以实现延迟调用？

interval join的缺点是只能输出关联上的结果，却无法输出未能关联上的结果（后续我需要对未关联上的结果进行特殊处理） 在 2022-12-07 13:33:50，"Lincoln Lee" 写道： >双流 join 的场景下可以考虑关联条件引入时间属性，变成 interval join 可以实现一定的时间对齐（ >https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql/queries/joins/#interval-joins >）

Re:Re: 如何扩展flink sql以实现延迟调用？

谢谢你给的建议，不过我们还没有升级到flink 1.16，目前在使用的是flink 1.15。 如果要使用flink sql来实现的话，是不是可以利用窗口去重来达到数据延迟关联的效果？ 在每条数据到达后开一个10分钟累加窗口(step和size均为10分钟）根据key去重，在等待窗口结束之时输出的去重结果再跟维表进行lookup join 在 2022-12-07 13:33:50，"Lincoln Lee" 写道： >双流 join 的场景下可以考虑关联条件引入时间属性，变成 interval join 可以实现一定的时间对齐（

(无主题)

退订

Re: 1.15.2作业频繁（每 几十分钟 ~ 1小时）报 LocalTransportException: readAddress(..) failed: Connection timed out .

谢谢，不过这几个参数和netty关系不大吧。 heartbeat和akka的可能会和rpc超时有关，不过我这个是netty的报错，不是rpc部分。 web和rest应该是和client提交任务有关。 Stan1005 <532338...@qq.com.invalid> 于2022年12月7日周三 15:51写道： > > 我也遇到过，tm的slot数一直是2，并行度高了就很容易出这个报错。tm内存保持为20480mb，相同的job讲并行度降低到256就没有报过这个。 > 另外可以考虑适当增加这几个参数（具体需要改动哪些建议先搜下这些参数的作用） > set

Re: 如何扩展flink sql以实现延迟调用？

双流 join 的场景下可以考虑关联条件引入时间属性，变成 interval join 可以实现一定的时间对齐（ https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql/queries/joins/#interval-joins ） 另外如果可以使用 lookup join 单边驱动关联并且不是所有数据都需要等待的话，可以尝试 lookup join 的延迟重试

flink on native k8s模式调度能否根据节点磁盘和网络io指标进行调度？

flink on native k8s模式调度能否根据节点磁盘和网络io指标进行调度？ 貌似现在只能根据cpu/内存剩余量进行调度，但如果新加一个节点会导致新作业全部的pod都部署到该节点上，造成该节点网络或磁盘IO飙升，这种情况有什么好的对策么？

Re:如何扩展flink sql以实现延迟调用？

有人能够解答一下吗？ 在 2022-11-26 11:20:34，"casel.chen" 写道： >双流关联场景下，流A数据到达后并不马上和流B进行关联，而是经过设置一段时间后再关联，这种场景下如何用flink >sql实现？如果当前不支持，需要怎样扩展flink sql呢？

flink sql是否支持延迟lookup join?

维表流数据晚于主表流数据到达甚至可能到达不了，所以想设置个5分钟等待窗口，关联上正常处理，关联不上发到另一个kafka topic，这种场景使用flink sql要如何实现？

flink web ui 提交任务报错

各位，问一下我一个main方法里面用了datastream api和table api，用命令行提交任务是没问题的，insert作业和其他的datastream job会拆开生成多个作业。但是当我用flink web ui提交任务时报上面这个错，有什么解决方案吗？ 我的代码： 中间还有一些processfunction

Re:flink on k8s节点网络io飙高问题如何解决？

flink on native kubernetes如何使用 affinity 配置软互斥？即同一个作业的不同pod分布在不同的节点node上 在 2022-12-05 19:51:02，"casel.chen" 写道： >我司flink作业运行在k8s集群上，日前发现有一些k8s集群节点的网络io在某些时间段超过了告警阈值180MB/s，最多达到430MB/s，最少的只有4MB/s，导致新作业无法部署到网络负载高的节点上，哪怕cpu和内存还有很多剩余。

1.15.2作业频繁（每 几十分钟 ~ 1小时）报 LocalTransportException: readAddress(..) failed: Connection timed out .

如题，这个问题长期存在，我想了解几个点： （1）connection time out 是连接时才会报的错误嘛？作业正常运行期间可能有嘛？我理解是连接时的报错，但是我看部分报错是作业运行不少时间才报错的（比如40分钟，1小时多），这种时刻为什么会有 connect 操作呢？netty的connection不是在作业启动时，就发 partition request 的时候创建好的嘛。 （2）之前调整过 netty 的 server 的 backlog，目前设置2048，不应该是这个导致。 （3）之前我TM都是1个slot，netty的server

flink executeAsync()

请问下 flink executeAsync() 一般都什么使用场景 无限流情况下executeAsync() / execute()是一样的吧 主要是想问下 场景/区别

flink sql滑动窗口问题

hello, all 我通过flink sql去group by uid计算滑动窗口，窗口大小一天，五分钟一个步长。譬如我uid有100w，那么五分钟就会往下游发送100w数据，如果是upsert-kafka，可能会发送200w数据（DELETE+INSERT）。 已经确定的是，大部分uid每五分钟计算出来的指标与前五分钟的一致，可能一天下来变化的uid只有几十w，有没有可能可以做一个判断，将要输出的指标与上一次的指标做一下判断，如果一致则不输出，如果不一致则输出？ 可以考虑不用滑动窗口，但是每个uid变化后需要输出到下游，不变化则不需要输出。 Best wishes

flink on k8s节点网络io飙高问题如何解决？

我司flink作业运行在k8s集群上，日前发现有一些k8s集群节点的网络io在某些时间段超过了告警阈值180MB/s，最多达到430MB/s，最少的只有4MB/s，导致新作业无法部署到网络负载高的节点上，哪怕cpu和内存还有很多剩余。 目前我想的办法是利用节点亲和性手动从负载高的节点上迁移出那些耗网络io高的作业pod到负载低的节点，但是过一段时间又会出现类似的问题，请问： 1. 有什么办法可以彻底消除这种网络负载不均衡问题么？ 2. k8s能否根据pod网络io负载进行合理调度吗？

Re:回复：flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？

好吧，难怪我没找到设置开始消费位置的参数，谢谢！ 在 2022-12-05 18:34:49，"JasonLee" <17610775...@163.com> 写道： >hi > > >Upsert-kafka 不支持指定消费者位置，默认是从 earliest 位置开始消费的，你可以自己修改代码支持 scan.startup.mode 参数。 > > >Best >JasonLee > > > 回复的原邮件 >| 发件人 | casel.chen | >| 发送日期 | 2022年12月5日 18:24 | >| 收件人 |

回复：flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？

hi Upsert-kafka 不支持指定消费者位置，默认是从 earliest 位置开始消费的，你可以自己修改代码支持 scan.startup.mode 参数。 Best JasonLee 回复的原邮件 | 发件人 | casel.chen | | 发送日期 | 2022年12月5日 18:24 | | 收件人 | user-zh@flink.apache.org | | 主题 | flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？ | flink

flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？

flink sql消费upsert-kafka源表如何指定从哪个位点开始消费？仿照kafka source表添加了 scan.startup.mode 参数会报非法参数

sql查询数据库不走索引

各位好！ 目前有一使用flink-sql编写的作业，其中存在通过jdbc查询mysql中某张表A需求，A表“b字段”为索引字段，但是flink-sql查询无法走到该表索引查询，为全表扫描查询。 代码类似于 CREATE TABLE A ( b decimal(4, 0), ..., ... ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://100.191.200.10:/lldb', 'username' = 'test', 'password' = 'Test!123',

退订

回复: 回复: Re: 怎样从flink执行计划json生成StreamGraph?

怎么指定？ 文档好像没有。sql的算子语义和DataStream不是一对一的，现在只看到Sink可以配置并行度。 从 Windows 版邮件发送 发件人: casel.chen 发送时间: 2022年12月2日 12:45 收件人: user-zh@flink.apache.org 主题: Re:回复: Re: 怎样从flink执行计划json生成StreamGraph? sql作业最终也会转换成stream api生成jobgraph，因此同样可以支持修改每个算子并行度 在 2022-11-30 11:24:50，"仙路尽头谁为峰" 写道：

Re: Re: flink sql作业无缝升级问题

通过savepoint方式先停止作业可以，不停止，你要考虑是否你的作业是否能做到重复处理部分数据不影响准确性。 先做savepoint但不停止作业，新作业启动后，新旧作业是消费的数据是重复的，不会因为相同group就不重复。 因为kafka的消费是2个模式，一个是组模式，还有一个是不受到组约束的。Flink采用的是后者。 我说的那个方法是在kafka后边加一个filter，filter的参数就是start和end，根据start和end过滤数据。 而且这个start和end需要可动态配置，就是不重启作业能配置才行。

Re:回复: Re: 怎样从flink执行计划json生成StreamGraph?

sql作业最终也会转换成stream api生成jobgraph，因此同样可以支持修改每个算子并行度 在 2022-11-30 11:24:50，"仙路尽头谁为峰" 写道： >Sql作业好像不支持修改每个算子并行度吧，修改并行度需要从头开始重新生成JobGraph提交作业。 >Json主要是贴到Plan Visualizer 开发和调试用。 >https://flink.apache.org/visualizer/ >从 Windows 版邮件发送 > >发件人: yidan zhao >发送时间: 2022年11月30日 10:12 >收件人:

Re:Re: flink sql作业无缝升级问题

拿kafka source作业为例，新老作业使用相同的consumer group，老作业先做savepoint，完了在老作业还在运行的同时启动新作业从刚才的savepoint恢复会有问题么？ 如何设置一个流量开关让新作业“准备”好再打开流量呢？有没有具体实操的例子？还是说需要自己修改flink源码，具体要看哪一个类方法？ 在 2022-11-30 20:08:44，"Jiangang Liu" 写道： >Flink目前无法做到无缝升级，需要走stop-with-savepoint、start

Re:FLINK postgresql CDC 报语法错误

是查询的时候报错,建表是成功的。 select * from postgres_cdc_test 错误： [ERROR] Could not execute SQL statement. Reason: org.postgresql.util.PSQLException: ERROR: syntax error 在 2022-12-02 10:09:37，"bmw" 写道： HI flink postgresql CDC flink1.12 ,postgresql:9.6.21 报错： CREATE TABLE

FLINK postgresql CDC 报语法错误

HI flink postgresql CDC flink1.12 ,postgresql:9.6.21 报错： CREATE TABLE postgres_cdc_test ( id INT, name STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'postgres-cdc', 'hostname' = '192.168.1.101', 'port' = '5432',

????

Re: flink sql作业无缝升级问题

应该是不行，必须先停止。 除非业务层面做了改动，像我业务的话我支持动态配置消费数据的开始结束时间的过滤。这样假设当前作业为A，当前时间9点55分，先动态设置A消费到10点就停止。在10点前启动新作业B，并设置作业B从10点的作业开始消费。这样10点之后比如10点5分左右确认作业A已经消费完10点前数据且sink完就可以停止了。 否则没办法，指定相同group也不可以的应该，我记得flink是使用主动assign分区的方式使用kafka的，因此如果前后作业同时存在，实际是重复消费，不存在共享消费的概念。 刘超 于2022年12月1日周四 09:36写道： > > kafka

Re: flink sql作业无缝升级问题

Flink目前无法做到无缝升级，需要走stop-with-savepoint、start job的流程，但是在这之间可以做一些优化来缩短恢复时间。比如，把新作业先启动起来，申请好资源，同时停掉老作业，将做好的savepoint用来触发新作业的执行。 casel.chen 于2022年11月29日周二 08:38写道： > 线上有一个流量较大的flink sql作业需要升级添加业务字段，当前主要是kafka (canal) 多表关联写入 mongodb > 数据同步场景，除了source offset外无其他状态，如何让用户对升级无感呢？ >

ElasticsearchSink 设置es 主分片数

hi， flink streaming（版本1.12.5） 写es的时候ElasticsearchSink.Builder发现没有设置配置的地方，比如要想设置 number_of_shards。哪位大佬知道，请教一下！

回复: Re: 怎样从flink执行计划json生成StreamGraph?

Sql作业好像不支持修改每个算子并行度吧，修改并行度需要从头开始重新生成JobGraph提交作业。 Json主要是贴到Plan Visualizer 开发和调试用。 https://flink.apache.org/visualizer/ 从 Windows 版邮件发送 发件人: yidan zhao 发送时间: 2022年11月30日 10:12 收件人: user-zh@flink.apache.org 主题: Re: Re: 怎样从flink执行计划json生成StreamGraph? 好吧，sql我具体不了解，我用的stream api比较多，我了解是stream

Re: Re: 怎样从flink执行计划json生成StreamGraph?

好吧，sql我具体不了解，我用的stream api比较多，我了解是stream api到streamGraph，然后到jobGraph，然后就是直接rest api方式提交给集群执行。 standalone场景。 casel.chen 于2022年11月30日周三 00:16写道： > > 如果要支持调整flink sql作业每个算子资源和并行度的话，不是要从json转回streamGraph再提交吗？ > > > > > > > > > > > > > > > > > > 在 2022-11-29 10:07:40，"yidan zhao" 写道： >

Re:Re: 怎样从flink执行计划json生成StreamGraph?

如果要支持调整flink sql作业每个算子资源和并行度的话，不是要从json转回streamGraph再提交吗？ 在 2022-11-29 10:07:40，"yidan zhao" 写道： >并不需要从执行计划json生成streamGraph呀~ >streamGraph提交之前直接转jobGraph。 > >casel.chen 于2022年11月28日周一 08:53写道： >> >> 源码中只找到如何从streamgraph生成执行计划的json串，但没有找到如何解析执行计划的json串生成回streamgraph，还请赐教

Re: FlinkCDC可以读到MySQL变更数据，但是插不到新的MySQL表里

> On Nov 4, 2022, at 2:34 PM, 左岩 <13520871...@163.com> wrote: > > tenv.executeSql("xxx); > env.execute(); 这样使用是不对的，你可以看下这两个方法的java doc 祝好， Leonard

Re: flink sql接cdc数据源按最新数据统计问题

> On Nov 29, 2022, at 8:32 AM, casel.chen wrote: > > 业务需求是mysql订单表按天按供应商实时统计交易金额，订单表会发生修改和删除，用flink > sql要如何实现呢？开窗取最新一条记录再聚合吗？如果遇到delete记录会不会减去相应的price呢？试着写了如下flink sql不知道对不对 会的，可以看下flink sql相关的原理文章，百度/谷歌一搜一大把。 祝好 Leonard > > > select > s.biddate, > s.supplier, > sum(s.price) >

Re: 如何使用flink sql优雅的处理大量嵌套if-else逻辑

我会选择 UDF + 配置文件，把配置文件放 HDFS上，UDF读这个配置文件。每次更新HDFS的配置文件，重启下任务 casel.chen 于2022年11月24日周四 12:01写道： > 我有一个flink > sql作业需要根据不同字段值满足不同条件来设置另一个字段值，还有一些嵌套if-else逻辑，这块逻辑不是固定的，业务方会过一段时间调整一次。 > 想请问如何使用flink sql优雅的处理嵌套if-else逻辑呢？我有想到使用drools规则引擎，通过udf来调用，不知道还有没有更好的办法？ > >

Re: 怎样从flink执行计划json生成StreamGraph?

并不需要从执行计划json生成streamGraph呀~ streamGraph提交之前直接转jobGraph。 casel.chen 于2022年11月28日周一 08:53写道： > > 源码中只找到如何从streamgraph生成执行计划的json串，但没有找到如何解析执行计划的json串生成回streamgraph，还请赐教

flink sql有办法获取到rowkind元数据字段吗？

flink sql有办法获取到rowkind元数据字段吗？比如按rowkind进行case when处理或者过滤

flink sql作业无缝升级问题

线上有一个流量较大的flink sql作业需要升级添加业务字段，当前主要是kafka (canal) 多表关联写入 mongodb 数据同步场景，除了source offset外无其他状态，如何让用户对升级无感呢？ 常规的停止作业再启动作业至少要几十秒，会造成消息积压告警，有没有可能先启新作业待运行平稳后再停止老作业？kafka group使用同一个，作业启动从group-offsets开始可以吗？另外，如果是有大状态作业又要如何无缝升级？

flink sql接cdc数据源按最新数据统计问题

业务需求是mysql订单表按天按供应商实时统计交易金额，订单表会发生修改和删除，用flink sql要如何实现呢？开窗取最新一条记录再聚合吗？如果遇到delete记录会不会减去相应的price呢？试着写了如下flink sql不知道对不对 select s.biddate, s.supplier, sum(s.price) from ( select * from ( select biddate, supplier, price,

Flink Kubernetes Operator何时会保留JobManager

环境: Flink：1.15 Operator: 1.2.0 问题： 注意到Operator源码里execution.shutdown-on-application-finish参数被设置为False。 想请问Flink operator部署的Flink作业在何时会保留JobManager的pod？因为我想在作业失败或者停止后还能获取到作业的日志。 除了正常Running状态会保留外，我尝试了其他情况。

回复：请问flink metrics如何获取任务状态？

hi 你的图片挂了 可以用图床上传一下图片 在这里贴个链接 另外自定义 reportor 把需要的metric 过滤出来上报就行了 回复的原邮件 | 发件人 | 陈佳豪 | | 发送日期 | 2022年11月28日 00:54 | | 收件人 | user-zh | | 主题 | 请问flink metrics如何获取任务状态？ | 自定义了一个kafka Metric Reporters #请问如何使用上述指标呢？ 我想通过上报获取任务状态。除了上述指标外如果有其他方案也可以，当前flink 版本是15.2 还望大神指教一番。

Flink Kubernetes Operator何时会保留JobManager Pod

环境: Flink：1.15 Operator: 1.2.0 问题： 注意到Operator源码里execution.shutdown-on-application-finish参数被设置为False。 想请问Flink operator部署的Flink作业在何时会保留JobManager的pod？因为我想在作业失败或者停止后还能获取到作业的日志。 除了正常Running状态会保留外，我尝试了其他情况。

请问flink sql可以被捕获异常吗？

hi 请问能在java代码里面try catch到基于flink sql写的任务异常信息吗？

Re:Re: Flink Kubernetes Operator何时会保留JobManager Pod

hi.如果是流作业呢？其实我是想在作业失败的时候能保留日志方便查看排查。 -- Best， Hjw 在 2022-11-28 15:33:37，"Biao Geng" 写道： >hi，主要就是针对作业FINISHED或者FAILED时也能保留作业。你可以跑一个批作业试试。 >Best， >Biao Geng > >获取 Outlook for iOS > >发件人: hjw >发送时间: Monday, November 28, 2022

Re:Re: 回复: FlinkCDC可以读到MySQL变更数据，但是插不到新的MySQL表里

CDC是自己编译的2.3，对应flink1.14的版本，还有一个问题是，可以读到变更数据。比如11点30写入mysql，flinkcdc读出来要慢几分钟，5~7分钟之后才能读到新写入或者变更的数据，第二个问题就行，变更数据插不到另外一张mysql表里 在 2022-11-07 10:11:56，"Shengkai Fang" 写道： >你用的是哪个版本的 cdc，增量部分的数据需要全量的部分读完才能进行。 > >Best, >Shengkai > >左岩 <13520871...@163.com> 于2022年11月4日周五 17:58写道： > >>

如果一个hive 数据库同时有hudi表和parquet，需要注册两个catalog？

如果一个hive 数据库同时有hudi和parquet表，好像只能分别注册hive catalog，和hudi catalog，两不同表需要分别使用通过catalog。不是很优雅，也对用户造成困惑。 select * from hudi_catalog.dbName.table1 left jon hive_catalog.dbBane.table2 spark 如果把hudi catalog注册名为spark_catalog，spark_catalog 是spark 默认cataog 名。sql 同时方式hudi 和parquet表，不需要添加catalog name。 select

Re: Flink Kubernetes Operator何时会保留JobManager Pod

hi，主要就是针对作业FINISHED或者FAILED时也能保留作业。你可以跑一个批作业试试。 Best， Biao Geng 获取 Outlook for iOS 发件人: hjw 发送时间: Monday, November 28, 2022 3:13:56 PM 收件人: user-zh@flink.apache.org 主题: Flink Kubernetes Operator何时会保留JobManager Pod 环境: Flink：1.15

Flink Kubernetes Operator何时会保留JobManager Pod

环境: Flink：1.15 Operator: 1.2.0 问题： 注意到Operator源码里execution.shutdown-on-application-finish参数被设置为False。 想请问Flink operator部署的Flink作业在何时会保留JobManager的pod？因为我想在作业失败或者停止后还能获取到作业的日志。 除了正常Running状态会保留外，我尝试了其他情况。

Re:回复：请问flink metrics如何获取任务状态？

hi 不好意思刚刚图好像又挂了 不知道这个能否查看。 在 2022-11-28 13:50:37，"m17610775726_1" 写道： hi 你的图片挂了 可以用图床上传一下图片 在这里贴个链接 另外自定义 reportor 把需要的metric 过滤出来上报就行了 回复的原邮件 | 发件人 | 陈佳豪 | | 发送日期 | 2022年11月28日 00:54 | | 收件人 | user-zh | | 主题 | 请问flink metrics如何获取任务状态？ | 自定义了一个kafka Metric Reporters

Re:回复：请问flink metrics如何获取任务状态？

这个metrics 我获取不到。 不知道要怎么配置才可以获取到。 在 2022-11-28 13:50:37，"m17610775726_1" 写道： hi 你的图片挂了 可以用图床上传一下图片 在这里贴个链接 另外自定义 reportor 把需要的metric 过滤出来上报就行了 回复的原邮件 | 发件人 | 陈佳豪 | | 发送日期 | 2022年11月28日 00:54 | | 收件人 | user-zh | | 主题 | 请问flink metrics如何获取任务状态？ | 自定义了一个kafka Metric Reporters

回复：请问flink metrics如何获取任务状态？

有大佬告诉下吗？ 这个指标的值获取不到。 | | 陈佳豪 邮箱：jagec...@yeah.net | 回复的原邮件 | 发件人 | 陈佳豪 | | 发送日期 | 2022年11月28日 00:54 | | 收件人 | user-zh | | 主题 | 请问flink metrics如何获取任务状态？ | 自定义了一个kafka Metric Reporters #请问如何使用上述指标呢？ 我想通过上报获取任务状态。除了上述指标外如果有其他方案也可以，当前flink 版本是15.2 还望大神指教一番。

怎样从flink执行计划json生成StreamGraph?

源码中只找到如何从streamgraph生成执行计划的json串，但没有找到如何解析执行计划的json串生成回streamgraph，还请赐教

请问flink metrics如何获取任务状态？

自定义了一个kafka Metric Reporters #请问如何使用上述指标呢？ 我想通过上报获取任务状态。除了上述指标外如果有其他方案也可以，当前flink 版本是15.2 还望大神指教一番。

如何扩展flink sql以实现延迟调用？

双流关联场景下，流A数据到达后并不马上和流B进行关联，而是经过设置一段时间后再关联，这种场景下如何用flink sql实现？如果当前不支持，需要怎样扩展flink sql呢？

Flink jdbc base-url 格式限制探讨

JdbcCatalogUtils 简单校验url 格式，对于pg，sqlserver 等数据库是有多catalog，且不同catalog 之间是不能相互访问，参考presto 设计处理，对于同一个pg实例的不同catalog，需要配置多个，具体到catalog，建议配置url 应该指定catalog。还有oracle数据，是需要指定sid，定制OracleCatalog 就遇到困惑。Spark jdbc 没有限制。 [image: image.png] 第二问题： oracle，pg，sqlserver Dialect中 quoteIdentifier方法，没有添加

[ANNOUNCE] Apache Flink 1.15.3 released

The Apache Flink community is very happy to announce the release of Apache Flink 1.15.3, which is the third bugfix release for the Apache Flink 1.15 series. Apache Flink® is an open-source stream processing framework for distributed, high-performing, always-available, and accurate data streaming

Re: 支持oracle、sqlserver、db2 jdbc catalog

https://github.com/melin/flink-cdc-catalog 准备补全jdbc catalog和cdc catalog melin li 于2022年11月24日周四 19:08写道： > flink jdbc catalog 只支持mysql pg，有计划支持oracle、sqlserver、db2 数据库？ >

flink-kafka-connector 消费时获取不到topic-paitition

kafka connector 开了这个配置, 'properties.allow.auto.create.topics' = 'true' 文档里面也有提到 , 但是开启flinkKafkaComsumer消费一个新的topic时，还是报找不到topic的错误，有大佬帮忙解释一下吗？ 报错如下： 这是我的配置 kafka broker 也开启了自动创建topic的配置

Re: Pyflink提交

Hi, 根据报错的提示，执行命令./python3.6.8.zip/bin/python3时没法导入pyflink，你可以在本地检查一下你的这个虚拟环境是不是没有成功安上pyflink Best, Xingbo 程龙 <13162790...@163.com> 于2022年11月25日周五 16:02写道： > 在使用pyflink提交任务时，部署模式onyarn > 1 在不使用Map等算子下如下参数 能够提交成功 并且运行 > .flink run -ynm pytest -m yarn-cluster -pyclientexec ***/python3 >

Re: Pyflink提交

集群端的 Python 环境中没有安装 PyFlink： ***/python3 这个环境 On Fri, Nov 25, 2022 at 4:02 PM 程龙 <13162790...@163.com> wrote: > 在使用pyflink提交任务时，部署模式onyarn > 1 在不使用Map等算子下如下参数 能够提交成功 并且运行 > .flink run -ynm pytest -m yarn-cluster -pyclientexec ***/python3 > -pyexec ***/python3 -pyarch *** /python3.6.8.zip

flink 1.15.3 已发布，但 maven repo 库中，暂无 1.15.3 的依赖

Pyflink提交

在使用pyflink提交任务时，部署模式onyarn 1 在不使用Map等算子下如下参数 能够提交成功 并且运行 .flink run -ynm pytest -m yarn-cluster -pyclientexec ***/python3 -pyexec ***/python3 -pyarch *** /python3.6.8.zip -py demo.py 2 在使用到map算子时 提交没有问题，但是运行报错，报错日志如下： .flink run -ynm pytest -m yarn-cluster -pyclientexec

Re: debezium-json数据timestamp类型时区问题

你在Oracle 数据库中的数据类型是TIMESTAMP 还是 TIMESTAMP WITH LOCAL TIME ZONE？ 我猜是后者，如果是后者直接在Flink SQL 里TIMESTAMP_LTZ 类型去映射就可以了 Oracle 的TIMESTAMP LTZ 类型和Flink SQL的TIMESTAMP LTZ类型含义和存储都是一致的语义，即epoch mills，存储时不需要时区。这两个类型都是在各自的系统中在在需要查看这些数据时，需要用 session 时区从epoch mills 转换成可读timestamp格式的字符串。 Oracle 设置session

Re: flink 新特性

hi，你可以访问这里 https://flink.apache.org/blog/ 也可以访问 apache flink 公众号 guanyq 于2022年11月24日周四 19:18写道： > 请问flink每个版本的新特性在哪里有介绍. -- Best regards

<    9   10   11   12   13   14   15   16   17   18   >