Re: flink1.16 sql gateway hive2

方勇老师说的没错。我们在文档里面也加了如何配置 hiveserver2 endpoint 的文档[1] [1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/hive-compatibility/hiveserver2/#setting-up Shammon FY 于2023年3月27日周一 08:41写道： > Hi > > >

Re: flink sql接cdc数据源关联维表写入下游数据库发现漏数据

听上去像是数据乱序了。可以看看这个文档对应的解决下[1] [1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/concepts/determinism/ Best, Shengkai casel.chen 于2023年3月1日周三 16:18写道： > flink sql上游接kafka canal json topic消费mysql同步过来的变更，接着关联几张维表，最后写入下游数据库发现漏数据。 > >

Re: flink sql jdbc connector是否支持多流拼接？

hi. 手动使用 join 将多个流拼接起来？ Best, Shengkai casel.chen 于2023年3月2日周四 21:01写道： > flink sql jdbc connector是否支持多流拼接？ > 业务上存在基于主键打宽场景，即多个流有相同的主键字段，除此之前其他字段没有重叠，现在想打成一张大宽表写入mysql/oracle这些关系数据库。 > 每条流更新大宽表的一部分字段。

Re: flink canal json格式忽略不识别的type

Hi. 能同时分享下复现这个 case 的sql 以及相关的报错栈吗？ Best, Shengkai casel.chen 于2023年2月9日周四 12:03写道： > 不同云厂商的数据同步工具对于全量+增量同步mysql数据到kafka canal json格式时行为不一致 > 有的厂商会将DDL语句同步到topic导致下游flink sql作业不能识别抛错，建议flink canal > json格式解析时直接忽略不识别的type，例如 > 例1： >

Re: [急] flink sql写数据到带主键表时能确保相同主键的记录都被同一个taskmanager处理吗？

我理解如果sink 表上带有 pk，就会主动 keyby 的[1]。 Best, Shengkai [1] https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java#L188 Shammon FY 于2023年2月20日周一 08:41写道： > Hi > >

Re: sql-client.sh 终止后, 进程不会被 kill

能 jstack 看看卡在哪里吗？并且提供下步骤该怎么复现这个。 Best, Shengkai <704669...@qq.com.invalid> 于2023年1月12日周四 22:03写道： > Hi 各位, > > > > 碰到sql-client进程不会被kill, 如下所示 > > --- > > Flink SQL> > > [4]+ Stopped ./bin/sql-client.sh > > [l3@node1 flink-1.16.0]$ jps > > 35040

Re: 如何扩展flink sql以支持CTAS/CDAS语句？

想问一下你想实现的功能是咋样的呢？是阿里云上的那种吗？ Best, Shengkai casel.chen 于2022年11月23日周三 08:29写道： > flink sql社区版貌似还不支持CTAS/CDAS语句，请问要如何扩展flink > sql以支持CTAS/CDAS语句？可以给个思路或代码示例吗？谢谢！

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

你用的是哪个版本的 cdc，增量部分的数据需要全量的部分读完才能进行。 Best, Shengkai 左岩 <13520871...@163.com> 于2022年11月4日周五 17:58写道： > > > > > > > > > > .print(); 去掉也不行， > > 跟这个也没有关系（一个tableEnv要执行2个DML），而且现在有一个现象，就是现在改了mysql表数据，flinkcdc得等好久才能读到变化数据，这是什么情况呢 > > > > > > > > > 在 2022-11-04 16:52:08，"yinghua...@163.com" 写道： > >

Re: upsert kafka作为source时，消费不到kafka中的数据

hi, 看不到的图片。能不能直接展示文字或者用图床工具？ Best, Shengkai 左岩 <13520871...@163.com> 于2022年10月28日周五 18:34写道： > upsert kafka作为source时，消费不到kafka中的数据 > 通过flink sql 写了一个upsert kafka connector 的连接器，去读一个topic,读不到数据，但是普通kafka > 连接器消费这个topic 就能读到，都是读的同一个topic，代码如下 >

Re: Re: Flink SQL 问题请教

我看到这个 sql 是对维表的join on字段进行 cast，想问一下能否对主表（X_NEWS_TCRNW0003_1_ALL_CDC）的字段进行 cast 试试呢？ Shengkai Alibaba yinghua...@163.com 于2022年10月24日周一 09:27写道： > 出错时SQL如下： > create table X_NEWS_TCRNW0003_1_ALL_CDC > ( > ID bigint, > NewsCode bigint, > NewsDate timestamp, > ITCode2 string, > CompanyCode

Re: 关于Sql Gateway / Rest Endpoint的Session超时时间的配置

没事。欢迎试用，如果在使用过程之中遇到了问题，可以联系我。 Best, Shengkai TonyChen 于2022年10月20日周四 09:45写道： > 找到了，打扰了列位 > public static final ConfigOption > SQL_GATEWAY_SESSION_IDLE_TIMEOUT = > key("sql-gateway.session.idle-timeout") > .durationType() >

Re: Re: flink cdc能否同步DDL语句?

Hi. 可以从这个地方入手看看 https://github.com/ververica/flink-cdc-connectors/blob/master/flink-connector-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/reader/MySqlRecordEmitter.java#L95 Best, Shengkai casel.chen 于2022年10月11日周二 10:58写道： > 可以给一些hints吗？看哪些类? > > > > > > > > > > > > > >

Re: 退订

你好，可以发送邮件到 user-zh-unsubscr...@flink.apache.org 来退订。 Best, Shengkai 13341000780 <13341000...@163.com> 于2022年10月10日周一 18:21写道： > > 退订 > > > > > -- > 发自我的网易邮箱手机智能版

Re: flink sql生成执行图中GroupWindowAggregate算子数不符合预期

hi. 能展示下具体想要的plan 和实际的 plan 吗？ Best, Shengkai 明寒 于2022年7月1日周五 09:50写道： > > HI：​在flink1.12中，对于如下的Sql，生成的执行图中有两个GroupWindowAggregate算子，该如何调整Sql或者配置保证只生成一个GroupWindowAggregate算子 > CREATE TEMPORARY TABLE RawSource ( > `key` STRING, > `accessNum` INT, > `status` STRING, > rowTime

Re: flink1.13.2 -> 1.13.6 任务cpu从50%左右飙升到1000%，window算子故障。

Hi. 这种情况下可以用 jprofile 看看到底 cpu 花在哪里。你可以使用火焰图或者 jstack 看看具体的栈和使用。 Best, Shengkai yidan zhao 于2022年6月28日周二 10:44写道： > 目前现象如题。任务就是kafkaSource读取数据，简单过滤，然后window，然后输出到mysql。 > > 目前来看运行后1-2min后cpu开始异常，不是马上异常。 异常时候window算子busy为100%。 > window是event time window，配合自定义的 >

Re: 使用join+聚合时，checkpoint异常

hi. 这种情况下，最好查看一下是否发生了反压，同时看看日志之中是否有相关的异常信息。 Best, Shengkai amber_...@qq.com.INVALID 于2022年6月21日周二 09:43写道： > 您好！ > 我使用flink1.14.4，sqlserver-cdc-2.2.1，yarn-per-job模式提交任务； > 当我提交普通数据同步任务时，一切正常； > 当我提交JOIN+聚合任务时，checkpoint无法正常工作，具体表现为无任何checkpoint记录，且Task Managed > Memory使用率始终是100%；

Re: Re: Re:Re: Re: Flink 使用interval join数据丢失疑问

表的ttl相关日志。 > > > > > > > > 在 2022-06-15 11:42:19，"Shengkai Fang" 写道： > >> 我在sql inner join里设置的状态ttl为20s,为什么20s的状态保留数据要比interval join开分钟级别的数据还要准确 > > > >不合理的 watermark 设置在 interval join 就会导致丢数据。设置 ttl 情况下，如果某个 key > >的数据频繁访问情况下，那么这个数据就不会过期。 >

Re: Re:Re: Re: Flink 使用interval join数据丢失疑问

的state数据格式来存数据（ListState、MapState等），原则上是多长时间没更新数据之后，就清空过期数据，是按照的“处理时间”来处理的。 > > > > > >如果我有不对的地方，请指正我哈。 > > > > > > > > > >-- > > > >Best！ > >Xuyang > > > > > > > > > > > >在 2022-06-12 14:39

Re: Re: Flink 使用interval join数据丢失疑问

>>>>== Optimized Physical Plan == > >>>>>Calc(select=[customer_id, goods_id, id, order_status, shop_id, > parent_order_id, order_at, pay_at, channel_id, root_order_id, id0, row_num, > p_sp_sub_amt, display_qty, qty, bom_type]) > >>>>>+- Join(joinType=[InnerJoin], where=[=(id

Re: Flink 使用interval join数据丢失疑问

你好，能提供下具体的 plan 供大家查看下吗？ 你可以直接 使用 tEnv.executeSql("Explain JSON_EXECUTION_PLAN ").print() 打印下相关的信息。 Best, Shengkai lxk 于2022年6月10日周五 10:29写道： > flink 版本:1.14.4 > 目前在使用flink interval join进行数据关联，在测试的时候发现一个问题，就是使用interval > join完之后数据会丢失，但是使用sql api，直接进行join，数据是正常的，没有丢失。 > 水印是直接使用kafka

Re: Re: 实现SupportsFilterPushDown接口过程中遇到的问题

RL was > not found on this > server"，发现是md文档中链接的URL拼写错误造成的。我提了一个hotfix[3]，正好借助这个PR熟悉下贡献流程，辛苦老师帮忙Review下 > > > >[1] > https://flink.apache.org/contributing/code-style-and-quality-common.html#nullability-of-the-mutable-parts > >[2] "usage of Java Optional"

Re: 实现SupportsFilterPushDown接口过程中遇到的问题

Hi. 我记得 Jdbc Connector 实现了 ProjectionPushDown。你可以参考着实现。 xuyang 老师说的对，getScanRuntimeProvider 发生在 push down 之后。应该不会有你说的问题。另外，可以考虑贡献到社区[1]，我们也可以帮忙一起 review 下，帮忙解决你的问题？ Best, Shengkai [1] https://issues.apache.org/jira/browse/FLINK-19651 Xuyang 于2022年6月1日周三 23:47写道： > >

Re: flink key by 逻辑疑问

Hi. 会根据 key 的 hash 值分配到固定个数的 keygroup 之中的。简单来说，跟HashMap> 有点相似。金竹老师有一篇文章详细解释了[1]。 如果想看实现的话，可以从 KeyGroupStreamPartitioner 入手来看看 Table 层是怎么做的。 Best, Shengkai [1] https://developer.aliyun.com/article/667562 Peihui He 于2022年5月29日周日 11:55写道： > Hi, all > > 请教下大家，flink key by 后 使用process

Re: LinkedMap ClassCastException issue

Hi. Could you tell us the version of the Flink you are using? What's the version of commons-collections:commons-collections:jar when you compile the sql and the version in the cluster? It's possible you compile the sql and submit with the different version. I am not sure how you submit your

Re: Does kafka key is supported in kafka sink table

Hi. Yes. Flink supports to write the value to the Kafka record key parts. You just need to specify which column belongs to the key in the WITH blocks, e.g. ``` CREATE TABLE kafka_sink ( ... ) WITH ( `key.fields` = 'id' ); ``` [1]

Re: flink rest 接口部分数据不返回，返回空，状态200正常。

你好，图挂了，应该是需要图床工具。 另外，能否贴一下相关的异常日志呢？ Best, Shengkai yidan zhao 于2022年5月20日周五 10:28写道： > UI视图：[image: 1.png]. > > 网络视图： > [image: image.png] > > > 补充部分集群部署信息： > （1）flink1.13，standalone集群，基于zk做的HA。3 jm，若干tm。 > （2）jm的rest api开启了ssl，基于 nginx >

Re: flink sql无法读取Hive映射的HBase表

hi, 能从日志中拿到更多详细的日志吗？请同时分享下你的执行步骤？ Best, Shengkai 18579099...@163.com <18579099...@163.com> 于2022年5月9日周一 17:46写道： > 我有一部分表的数据是存在hbase上的，平时通过hive加载外部表的方式读取hbase的数据，我想通过flink sql读取hive表的方式 > 读取数据（不直接使用flink > 读取hbase是我使用的catalog是hive，不用再写建表语句然后再查），当我用sql-client尝试的时候报错。 > 读取正常的hive是可以正常读取的，但是读取hive

Re: CsvReaderFormat类的位置

Hi, 看不到你的图，是说这个吗[1]？ Best, Shengkai [1] https://github.com/apache/flink/blob/ca58a700bbc0522f3c62e9db720f9f89c8bd8313/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvReaderFormat.java 于2022年5月7日周六 10:07写道： > 您好 > >

Re: flink-connector和flink-sql-connector的区别

hi sql jar 往往是 shade 了相关的依赖，而 普通的 jar 则不带有相关的依赖。正如名字所说，在 table api/sql 的情况下建议使用 sql jar，datastream 建议使用 普通的jar。 Best, Shengkai weishishuo...@163.com 于2022年4月21日周四 16:52写道： > >

Re: (无主题)

退订的同学，请发送任意邮件到 user-zh-unsubscr...@flink.apache.org。 BEST, Shengkai 王健 <13166339...@163.com> 于2022年4月15日周五 14:40写道： > 退订

Re: 退订

退订请发送邮件到 user-zh-unsubscr...@flink.apache.org。 Best, Shengkai 剑来 <1370505...@qq.com.invalid> 于2022年4月15日周五 16:28写道： > 退订

Re: conf目录中flink配置文件参数

Hi, 这个就是 Flink UI 监听的端口号。 Best, Shengkai 陈卓宇 <2572805...@qq.com.invalid> 于2022年4月15日周五 17:47写道： > 今天看到这两个配置项，有些遗憾： > # Port range for the REST and web server to bind to. > # > rest.bind-port: 8080-8090 > 这个绑定端口是起到什么作用？ > # The address that the REST web server binds to > # > #rest.bind-address:

Re: 退订

退订的同学，请发送任意邮件到 user-zh-unsubscr...@flink.apache.org。 Best, Shengkai yefan 于2022年4月15日周五 16:33写道： > 退订

Re: flink sink ES7 报错：Invalid lambda deserialization

退订的同学，请发送任意邮件到 user-zh-unsubscr...@flink.apache.org。 Best， Shengkai Shengkai Fang 于2022年4月15日周五 17:34写道： > 我记得这个问题常常是因为 lambda 函数的入参/出参被 shade 了，导致找不到对应的实现类了。不太理解你的意思，但是用 sql-client > 使用 sql-jar 应该是没有问题的。建议你代码里面也换成 sql jar 先试试看。 > > Best, > Shengkai > > wangzhen 于2022年4月1

Re: flink sink ES7 报错：Invalid lambda deserialization

我记得这个问题常常是因为 lambda 函数的入参/出参被 shade 了，导致找不到对应的实现类了。不太理解你的意思，但是用 sql-client 使用 sql-jar 应该是没有问题的。建议你代码里面也换成 sql jar 先试试看。 Best, Shengkai wangzhen 于2022年4月15日周五 11:57写道： > 退订 > > > > > > > > > > > > > > > > > > 在 2022-04-14 22:44:48，"顺其自然" <712677...@qq.com.INVALID> 写道： > >我的flink 1.12.1 sql

Re: 如何条件查询不扫描全表

Hi, 可以为 HBaseDynamicTableSource 实现 FilterPushDown[1] 接口。这样子 你直接添加的 filter 会被下推到 Source，从而在读取的时候过滤掉不相关的数据。 [1]

Re: Undeliverable: Re: flink1.13.1 sql client connect hivecatalog 报错

HGT+HCbsAz9Hv5RU > ruC/3H5UUhkL29DNCLPXsQx3U4qAbHNFIQtA1CvI99j55Pg= > X-Google-Smtp-Source: > ABdhPJx7/O7Ag3uEhwBNFyIWBHDPTxLpItw1QDymjd+Joj8OW3RpFtKmnbnsXaaxXsxC33L2O+Ki+LWm4/vME6cnu70= > X-Received: by 2002:a05:6602:13c9:: with SMTP id > o9mr19655702iov.40.1637547288489; &

Re: flink1.13.1 sql client connect hivecatalog 报错

你好，想问一下你的 hive 包确定是1.13对应的版本吗？ Best, Shengkai drewfranklin 于2021年11月22日周一 上午9:45写道： > Hello >我按照官方文档使用 sql client 去连接hive catalog 时出错。 > 我的hive version 2.3.6 > Flink version 1.13.1 > > 感觉官方介绍的bundled 方式添加jar 包，在flink/lib 下添加如下截图的包。然后重启集群，启动了sql-client >

Re: flink1.13.1 sql client connect hivecatalog 报错

hi, 看不见图，建议用图床或者填一下代码。 我看到代码中有 yaml 文件，事实上 更建议使用 ddl 来创建相应的 catalog。 best, Shengkai drewfranklin 于2021年11月18日周四 下午6:01写道： > Hello, friends ! >我按照官方文档使用 sql client 去连接hive catalog 时出错。 > 我的hive version 2.3.6 > Flink version 1.13.1 > > 感觉官方介绍的bundled 方式添加jar 包，在flink/lib

Re: 如何添加source保障程序状态不丢

不太理解，直接删掉老source 有什么影响吗？ Best, Shengkai 杨浩 于2021年11月17日周三 下午6:00写道： > 试了下，可以直接加source，有办法加source后，删掉老的source么？ > > > > > > > > > > > > > > > > > 在 2021-11-17 14:56:37，"杨浩" 写道： > > 请问下，我们程序在运行后需要添加一个kafka > source（和老的协议一致，后面加工逻辑一样），程序开启了checkpoint，如何保障之前程序的状态不丢啊 > > > > > >

Re: flink sql建表异常问题

打出来的jar中是否包含了 相应的依赖呢？ aiden <18765295...@163.com> 于2021年11月17日周三 下午2:34写道： > 补充下抛错异常 > Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: > Could not find a suitable table factory for > 'org.apache.flink.table.delegation.ExecutorFactory' in > the classpath. > > Reason: No

Re: Flink没有Operator级别的数据量Metrics

同一个 task 里面的 op 往往是 chain 起来的，不经过网络传输很难评估其传输的数据量的...如果是 debug 的话，应该把所有的 chaining 断开，这样子每个 op 就是一个 task。可以通过设置 pipeline.operator-chaining false 来关闭chaining。 Best, Shengkai Ada Luna 于2021年11月17日周三 上午10:16写道： > 看不到Task里Operator之间传输的数据量 > > zhisheng 于2021年11月4日周四 下午4:56写道： > > > > webui 有

Re: Flink jdbc Connector 特殊类型问题

如果是个 bug，建议在社区开个 issue 跟踪下这个问题。 Shengkai Fang 于2021年11月16日周二 下午12:37写道： > 能分享下具体是什么错误类型吗？ > > 我看了下代码，感觉不太好支持。具体的序列化器是由 > `AbstractJdbcRowConverter`#createExternalConverter 决定的。 > 根据你的描述，当前的序列化器不够通用导致了这个问题，可能需要改动下源码才能支持。 > > Best, > Shengkai > > Ada Luna 于2021年11月

Re: Flink jdbc Connector 特殊类型问题

能分享下具体是什么错误类型吗？ 我看了下代码，感觉不太好支持。具体的序列化器是由 `AbstractJdbcRowConverter`#createExternalConverter 决定的。 根据你的描述，当前的序列化器不够通用导致了这个问题，可能需要改动下源码才能支持。 Best, Shengkai Ada Luna 于2021年11月12日周五 上午11:25写道： > Oracle中有VARCHAR 和 CLOB > 如果我在Flink SQL JDBC Sink中配置STRING那么只能写VARCHAR 写CLOB会报错。 > 我想扩展FlinkSQL

Re: Re: FlinkSQL 1.12 Temporal Joins 多表关联问题

1. 不知道使用 udf 能不能实现，自己实现一个 udf，然后在实现里面手动查询外表； 2. 如果自己实现的话，那么也应该能控制攒 batch 的实现； 悟空 于2021年11月12日周五 上午11:53写道： > Hi : > 第一个 我了解了Cache 不太适合我的场景，因为我的表都是几十亿量级，同时 我要根据一些关键键 去数据库里查询，所以 > 我先在Job 中 聚合一些主键，通过In 条件 去查询。 > 第二个 好像是我理解的问题，最初想通过Flink Sql 把整体逻辑 下发到数据库去查询，因为有些OLAP > 引擎 查询性能是可以接受的 > > > > --- >

Re: flink new source api， kafka部分对kafka-client的版本要求。

我看pom 之中之前依赖的kafka的版本是 2.4.1，当前依赖的 kafka 版本是 2.8.1。应该是不支持低版本的 kafka。 Best, Shengkai yidan zhao 于2021年11月11日周四 下午3:22写道： > > 如题，当前新的kafaksouce貌似对kafka-client版本做了依赖，比如代码KafkaPartitionSplitReader.acquireAndSetStoppingOffsets方法中用到的 > > consumer.committed(partitionsStoppingAtCommitted) >

Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中

说的是 statement set [1] 吗 ？ [1] https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/sqlclient/#execute-a-set-of-sql-statements 悟空 于2021年8月26日周四 上午11:33写道： > hi all: > 我目前基于flink 1.12 sql 来开发功能， 目前遇到一个问题， 我现在想实现 在一个事务里 先将kafka > 源的数据写入到一张msyql 表中，

Re: Re: Flink任务假死；无限100%反压；但下游节点无压力。

- 得看一下具体的卡死的节点的栈，分析下具体的工作任务才知道。 - 日志中有包含错误的信息吗？ Best, Shengkai yidan zhao 于2021年8月26日周四 下午12:03写道： > 可能存在机器压力倾斜，但是我是不太清楚这种现象的原因，直接停滞了任务？ > > 东东 于2021年8月26日周四 上午11:06写道： > > > 建议检查一下是否有数据倾斜 > > > > > > 在 2021-08-26 10:22:54，"yidan zhao" 写道： > > >问题期间的确ckpt时间较长。 > >

Re: flink table over 窗口报错

able.operations.utils.OperationTreeBuilder.project(OperationTreeBuilder.java:183) > at > org.apache.flink.table.api.internal.TableImpl$OverWindowedTableImpl.select(TableImpl.java:956) > at com.rk.linkdata.flink.BatchFlinkTask.main(BatchFlinkTask.java:44) > > > > > Flink 版

Re: flink table over 窗口报错

能发一下具体的异常栈吗？是哪个版本？ yanyunpeng 于2021年8月4日周三 下午2:47写道： > Table table = tableEnv > .from("t_yyp_test") > .window(Over.partitionBy($("f_h"), $("f_l"), $("f_j")) > .orderBy($("f_time")) > .preceding("unbounded_range") > .following(CURRENT_RANGE) > .as("w")) > .select($("f_value"), >

Re: 回复：场景题：Flink SQL 不支持 INSERT INTO… ON DUPLICATE KEY UPDATE ?

Flink 暂时不支持这个功能，可能需要自己改一下 jdbc connector 相关的代码. 但是这个报错很奇怪..你 sql 咋写的 Ye Chen 于2021年8月2日周一 上午11:37写道： > 你好，我试了一下，如果表的ddl是三个字段，但是insert只指定两个字段的话，会报错： > [ERROR] Could not execute SQL statement. Reason: > org.apache.flink.table.api.ValidationException: Column types of query > result and sink for

Re: flink1.12 sql-client 临时文件

hi，想问一下是在什么情况下产生这些文件的？相关的 sql 是什么？ 如果在 tmp 下 似乎不是 sql client 产生的。 吴旭 于2021年7月30日周五 下午2:45写道： > > > > 大佬们，请问下 在使用 sql-client 查询的过程中， /tmp 目录下面生成很多如下这样的临时文件，请问有地方可以配置吗 > > > ```txt > 00615a2c-c0f6-4ca9-b5c4-ee8d69ca2513 > 1098b539-31f2-4fbb-9e7b-46d490ff25d6 > 13b0dcbb-2e2c-4b85-9969-f90915e2a9ca

Re: Flink1.13.1自定义Catalog问题出错

Hi. 新版的 CatalogFactory 实现了 Factory，这意味着当前的所有的 connector、format 以及 Catalog 都实现了相同的接口，保持了统一性。而保持原来的方法，更多是为了暂时的兼容性( 我的理解 )： 如果 某个Catalog 从低版本迁移到高版本只需要添加一些新的接口方法，而不需要删除之前的逻辑。之后的版本可能会删除这些已经被deprecated 方法。 Best, Shengkai. hoose 于2021年7月15日周四 下午6:13写道： > Flink1.13.1在Catalog方面修改比较大，特别是一些方法的实现上 > 如 >

Re: sql client提交 flink任务失败

可以看看之前的问题，看看能否解决。 Best, Shengkai [1] http://apache-flink.147419.n8.nabble.com/Flink-td7866.html [2] https://issues.apache.org/jira/browse/FLINK-20780 Fei Han 于2021年6月8日周二 下午8:03写道： > > @all: > Flink环境：Flink1.13.1 > HADOOP环境：CDH5.15.2 > 测试命令如下：./bin/sql-client.sh embedded -i

Re: flink 1.13.0 ，使用flink sql 链接数据库是否支持多模式，即表名为schema.name

请问是要用正则表达式匹配数据库中的table吗？‘org.users’ 是一个正则表达式吗？ Best, Shengkai Asahi Lee <978466...@qq.com> 于2021年5月19日周三 下午2:01写道： > hi! >flink jdbc 是否有考虑支持表基于模式查询？如下 table-name写法： > CREATE TABLE MyUserTable ( id BIGINT, name STRING, age INT, status > BOOLEAN, PRIMARY KEY (id) NOT ENFORCED ) WITH

Re: Upsert kafka 作为 source 的几个问题

Hi. 1. 当初的设计是一个较为保守的设计，其主要目的就是为了能够补全delete消息； 2. 核心类是 StreamExecChangelogNormalize[1] 3. 是的。目前 Upsert-kafka 要求具有相同key的数据在相同 partition 的。因为 kafka 仅保证 partiiton 内按 offset 读取，如果相同 key 的数据分布在不同 partition 的话，那么读取会乱序。 4. 当数据进入到具体的算子的时候并不会区别数据是来自什么connector的。如果 left, right 的 paritition 策略不一致，会shuffle的。

Re: Table-api sql 预检查

Hi. 可以看看 TableEnvironment#execute逻辑，如果能够将sql 转化成 `Transformation`，那么语法应该没有问题。 Best, Shengkai Michael Ran 于2021年4月29日周四 上午11:57写道： > dear all : > 用table api 提交多个SQL 的时候，有什么API 能提前检查SQL 的错误吗？ 每次都是提交执行的时候才报错。 > 理论上里面在SQL解析部分就能发现，有现成的API吗？还是说需要自己去抓那部分解析代码，然后封装出API。 >

Re: 使用Table API怎么构造多个sink

Hi. 可以通过`StatementSet` 指定多个insert，这样子就可以构造出多个sink了。 Best, Shengkai Han Han1 Yue 于2021年4月28日周三 下午2:30写道： > Hi, > 个人在分析RelNodeBlock逻辑，多个SINK才会拆分并重用公共子树，怎么构造多个sink呢， > 文件RelNodeBlock.scala源码里的writeToSink()已经找不到了 > > // 源码里的多sink例子 > val sourceTable = tEnv.scan("test_table").select('a, 'b, 'c) >

Re: flink sql 使用自定义函数 ，返回嵌套行， 查询报错 scala.MatchError

hi, gen. 近期内应该就会发布，应该是五一左右就会发布1.13的版本。 Best, Shengkai gen 于2021年4月27日周二 下午8:57写道： > hi, Shengkai > 非常感谢你的解答， 解决了困扰我几天的问题。 > 按照你的建议 ，我使用 今天（2021-4-27） 主干版本，运行正常，发现确实是已经修复的。 > 我之前使用的版本是 1.12.2。 > > > 目前最新的release版本是1.12.2 ，应该还没有包含这个修复。不知道你是否了解 1.13的发布计划。 > > > > -- > Sent from:

Re: flink sql 使用自定义函数 ，返回嵌套行， 查询报错 scala.MatchError

Hi gen 我在1.13分支上验证了下你的case，发现能够跑通。建议cp下那个patch到自己的分支，再验证下。 Best, Shengkai Shengkai Fang 于2021年4月27日周二 上午11:46写道： > 请问你使用的是哪个版本？ 这个似乎是一个已知的修复的bug[1] > > [1] https://github.com/apache/flink/pull/15548 > > gen 于2021年4月27日周二 上午9:40写道： > >> Hi, all >> >&

Re: 问题：flink 1.13编译 flink-parquet报错 -类重复：org.apache.flink.formats.parquet.generated.SimpleRecord

是不是没有删除之前生成的类，手动删除冲突的类试试。 Best, Shengkai HunterXHunter <1356469...@qq.com> 于2021年4月27日周二 上午10:58写道： > 查看发现 > > org.apache.avro > avro-maven-plugin > ${avro.version} >

Re: Flink消费Kafka报错：ByteArraySerializer is not an instance of org.apache.kafka.common.serialization.Serializer

hi, Colar. Flink 使用的 Kafka 的版本是2.4.1，但是你的集群版本是1.1.1。看样子 作业运行时加载的是 集群上的 ByteArraySerializer，而不是 Flink 的 `flink-connector-kafka`中的。不太确定打成一个shade包能不能行。 Best, Shengkai Colar <523774...@qq.com> 于2021年4月26日周一 下午6:05写道： > 使用Flink 1.12.2 消费Kafka报错： > > 2021-04-26 17:39:39,802 WARN

Re: flink sql 使用自定义函数 ，返回嵌套行， 查询报错 scala.MatchError

请问你使用的是哪个版本？ 这个似乎是一个已知的修复的bug[1] [1] https://github.com/apache/flink/pull/15548 gen 于2021年4月27日周二 上午9:40写道： > Hi, all > > 请教下为什么 无法通过t.* 将 自定义函数返回的嵌套字段查出来。 > > tEnv.executeSql( > """ > | SELECT t.* FROM ( > | SELECT EvtParser(request) as t FROM parsed_nginx_log >

Re: DataStreamAPI 与flink sql疑问

Flink支持将DataStream 转换成一个 Table，然后通过API进行操作。如果想跟SQL相结合，可以将Table注册成一个 temporary view。 Best, Shengkai HunterXHunter <1356469...@qq.com> 于2021年4月27日周二 上午9:46写道： > 你试过吗？ > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/ >

Re: Dynamic Table Options 被优化器去掉了

hi, macial kk. 看样子是个bug，能提供以下你的ddl以及相关的环境吗？方便我们复现下问题。 Best, Shengkai plan的digest是不会打印connector的option的值的，因此你是没有办法通过plan来判断是否生效了。 macia kk 于2021年4月26日周一 上午12:31写道： > Hi > > 我有在使用 temporal Joini 的时候有设置 如果读取分区的相关的 dynamic > option，但是最后是没有生效的，我看全部使用的默认参数，打印出来了执行计划，逻辑执行计划是有的，优化之后没有了 >

Re: 官网文档和样例的不完整性和不严谨性的问题

Hi, xuefli. 非常感谢你指出文档的问题！ 由于邮件中看代码比较吃力（没有语法高亮以及排版的问题），我只是粗略地看了下代码。 当输入源 为 `一次性从内存中的List读取数据`，无法触发onTimer。 实际的例子中，我看到看到采用的是process time，且延时 3s 触发 。我怀疑是不是，数据量太少，所以程序很快就结束了导致没来得及触发timer，建议改成event time试试这种情况。 Best, Shengkai xue...@outlook.com 于2021年4月25日周日 上午9:42写道： >

Re: 关于upsert-kafka connector的问题

这里有对upsert-kafka完整的一个分析的讲解：深度解析 Flink upsert-kafka[1]。如果还有问题，可以继续咨询。 [1]https://flink-learning.org.cn/developers/flink-training-course3/ Shengkai Fang 于2021年4月25日周日 上午10:16写道： > 本质上，upsert-kafka是对kafka的封装，其内部仍然是一个消息队列，只是在消费的时候，我们形成一个视图。 > > 消息从flink进入到kafka之中，根据kafka的协议保证了at-l

Re: 关于upsert-kafka connector的问题

本质上，upsert-kafka是对kafka的封装，其内部仍然是一个消息队列，只是在消费的时候，我们形成一个视图。 消息从flink进入到kafka之中，根据kafka的协议保证了at-least-once。 Best, Shengkai op <520075...@qq.com> 于2021年4月23日周五 下午2:18写道： > > 谢谢，upsert-kafka作为sink可以保证相同key的数据放在同一个partition内，假如对相同key的更新数据，由于网络等原因后更新的值A的比先更新的值B提前发送到kafka, >

Re: 关于upsert-kafka connector的问题

如果数据在upsert-kafka中已经做到了按序存储（相同key的数据放在同一个partition内），那么flink消费的时候可以做到保序。 Best, Shengkai

Re: 关于upsert-kafka connector的问题

Hi, 请问是有什么具体的问题吗？ Best, Shengkai op <520075...@qq.com> 于2021年4月22日周四 下午6:05写道： > 用 upsert-kafka connector 作为source，会有key的插入和更新出现乱序导致结果不准的问题吗？ > 谢谢

Re: 关于Flink水位线与时间戳分配的疑问

hi, 图挂了。 1. 可以这么使用这个方法： ··· input.assignTimestampsAndWatermarks( WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMillis(10)) .withTimestampAssigner((event, timestamp) -> 42L)); ··· TimestampAssigner 会从输入的event上读取数据 并由watermark generator 决定输出对应的watermark. 3.

Re: [讨论] Flink Connector 并行写入数据方案

Hi jie. User mail list 更多是用来讨论使用中的问题，请将关于dev相关的问题转发到d...@flink.apache.org 详情可以参考[1] [1] https://flink.apache.org/community.html jie mei 于2021年3月31日周三 下午3:03写道： > Hi, Community > > 我想发起一个初步的讨论，关于如何扩大 Flink 写入数据的并行度，并且能够分表，可选事务支持的方案。 > > 该方案应该支持三种场景： > > 1) 至少一次： 不支持或者有限支持 update 的存储，通常通过查询去重。

Re: Upsert Kafka 的format 为什么要求是INSERT-ONLY的

为是一次insert，从而促使下游emit record? > > 我们现在有若干自定义的Format，如果为了适配Upsert Kafka，format需要对d，(-u) 事件发射value == > null的Record吗 > > > > > > 发件人: Shengkai Fang > 发送时间: 2021年3月15日 14:21:31 > 收件人: user-zh@flink.apache.org > 主题: Re: Upsert Kafka 的

Re: Upsert Kafka 的format 为什么要求是INSERT-ONLY的

Hi. 当初的设计是基于kafka的compacted topic设计的，而compacted topic有自身的表达changelog的语法，例如：使用value 为 null 表示tombstone message。从这个角度出发的话，我们仅从kafka的角度去理解数据，而非从format的角度去解析数据。 这当然引入了一些问题，例如当利用upsert-kafka读取数据的时候需要维护一个state以记住读取的所有的key。 Best, Shengkai 刘首维 于2021年3月15日周一 上午11:48写道： > Hi all, > > > >

Re: 咨询求助

hi, 根据文档[1][2]， 你可以通过在with参数内填相应的内容来通过认证, e.g 'properties.sasl.kerberos.service.name' = 'xyz' [1] https://kafka.apache.org/documentation/#brokerconfigs_sasl.kerberos.service.name [2] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/kafka.html#properties 瞿叶奇

Re: flink sql 执行limit 很少的语句依然会暴增

orExecution(PackagedProgram.java:198) > at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:149) > at > > org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:699) > at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:232) > at > > org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:9

Re: flink sql 执行limit 很少的语句依然会暴增

hi, LIMIT PUSH DOWN 近期已经merge进了master分支了。 [1] https://github.com/apache/flink/pull/13800 Land 于2021年1月22日周五 上午11:28写道： > 可能是没有下推到MySQL执行。 > 问题和我遇到的类似： > http://apache-flink.147419.n8.nabble.com/Flink-MySQL-td10374.html > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/ >

Re: 回复： flink sql 消费kafka 消息 写入Hive不提交分区

这种情况一般是kafka的某个分区，不存在数据，导致总体的watermark不前进。遇到这种情况一般是需要手动设置idle source[1]。但是社区的watemark push down存在一些问题[2]，已经在修复了。 [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/config.html#table-exec-source-idle-timeout [2]

Re: Some questions about limit push down

hi, Jun Zhang. Currently, the rule `PushLimitIntoTableSourceScanRule` captures the structure that the `FlinkLogicalSort` node is the parent of the `FlinkLogicalTableSourceScan`. In your case, we have a Calc node between the Sort node and the Scan node, which makes the rule fails to apply. Maybe

Re: Flink 1.11.2客户端 select * from table

要看一下日志里面是否有报错信息，才能进一步分析。 Jacob <17691150...@163.com> 于2020年12月29日周二 上午9:11写道： > Dear All, > > > > > > 在Flink SQL客户端中使用select * from table 语句查询表数据，结果只显示表头，没有数据，是什么原因。 > >

Re: Flink SQL并发度设置问题

30，那么默认情况window的最大并行度是128。我在想，如果按照平均考虑，这种情况是不是从机制上就已经有大概率会导致数据倾斜了呢？设置成32对于128才可以均衡不是吗。 > > Shengkai Fang 于2020年12月27日周日 下午3:46写道： > > > 可以通过该配置[1]来设置 > > > > [1] > > > > > https://ci.apache.org/projects/flink/flink-docs-master/dev/table/config.html#table-exec

Re: 求问为什么KafkaDynamicSource 在批模式下不能运行

现在Flink 走的是流批一体， 为什么说 Kafka 不支持批模式呢？ wxpcc 于2020年12月25日周五 下午7:10写道： > kafka在我们这边场景上除了用来存放实时流式数据，还会用作临时大数据量的存储，主要用于： > > 1. > > 数据同步时，将全量数据同步到一个临时的kafka中，增量数据持续性同步到kafka中，目前我们都使用流模式消费其中的数据，就会有手动停止，或者借助指标等自动停止流式任务 > 2. 数据恢复时 > 3. 临时查看某个时间区间的数据 > > 如果批模式 sql能够完成这些事情的话那该多好 > > > > -- > Sent from:

Re: rowtime的时区问题

社区正在解决这个问题，1.13应该会有一个系统性地修复。 CC Leonard 作为work around，可以参考下这个博客[1] [1] https://blog.csdn.net/tzs_1041218129/article/details/109064015?utm_medium=distribute.pc_relevant.none-task-blog-title-3=1001.2101.3001.4242 ゞ野蠻遊戲χ 于2020年12月26日周六 下午11:16写道： > Hi 大家好 > >

Re: Flink SQL并发度设置问题

可以通过该配置[1]来设置 [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/config.html#table-exec-resource-default-parallelism 赵一旦 于2020年12月27日周日 下午12:44写道： > 了解下多少数据量呀，128的并发其实很高了感觉。 > > guaishushu1...@163.com 于2020年12月26日周六 下午5:39写道： > > > Flink > > >

Re: 【Flink的shuffle Mode】

Hi, 在Datastream api之中我们可以控制shuffle的模式[1]，但是在Table api 似乎暂时还不提供控制shuffle mode的接口。 我对join算子的实现不太熟悉，这个需要shuffle吗？对于agg操作，它的partition模式是由group by的key控制的。 [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/operators/#physical-partitioning 忝忝向仧 <153488...@qq.com> 于2020年10月6日周二

Re: 请问如何模拟一个back pressure的场景,应该如何实现?谢谢

hi, Jark 老师有个博文提到过这些，可以参考一下并模拟下[1][2]。 [1] http://wuchong.me/blog/2016/04/26/flink-internals-how-to-handle-backpressure/ [2] https://www.ververica.com/blog/how-flink-handles-backpressure 大森林 于2020年10月3日周六 下午8:03写道： > 您好! > 请问处理的速度 > 和消费的速度 > 在代码上应该怎么体现呢? > 谢谢您 > > >

Re: 含有多个kafka-source的job里如果用events_time作为时间的话，如何解决快流通过水印覆盖漫流的情况。

hi， 你说的是为每个partition生成一个watermark吗？ 这样子快流和慢流都会有独立的watermark gererator。 datastream已经支持了该特性， table层正在支持该特性，你可以看看flink-19282的jira。 赵一旦 于2020年9月28日 周一上午11:39写道： > 我这边负责的作业，一个作业上有2-3个kafka数据源，还包括多个mysql配置流数据源。也是各种join，但是没有union的case。 > > 没有任何watermark的问题，flink现有机制都是可以完美解决的。 > > > > 赵一旦 于2020年9月28日周一

Re: flink 1.9 关于回撤流的问题

hi, 我对于使用upsert kafka能够省state感到疑惑。金竹老师提供的实现只是丢掉了delete消息，你的下游表依旧需要手动去重才可以得到准确的结果才对啊。如果每个下游表都手动去重这样子还能省state吗？ star <3149768...@qq.com> 于2020年6月8日周一 上午9:38写道： > 非常感谢，正是我想要的。也谢谢金竹老师的分享！ > > > > > --原始邮件-- > 发件人:"Sun.Zhu"<17626017...@163.com; > 发送时间:2020年6月7日(星期天)

Re: TableColumn为啥不包含comment

hi, 那请你在那个jira留一下言，我会把这个分配给你。 Harold.Miao 于2020年8月17日周一 上午11:26写道： > 谢谢 我想提交这个patch > > Shengkai Fang 于2020年8月14日周五 下午4:33写道： > > > hi, 我已经建了一个issue[1]跟踪这个情况，有兴趣的话可以帮忙修复下这个bug。 > > > > [1] https://issues.apache.org/jira/browse/FLINK-18958 > > >

Re: TableColumn为啥不包含comment

hi, 我已经建了一个issue[1]跟踪这个情况，有兴趣的话可以帮忙修复下这个bug。 [1] https://issues.apache.org/jira/browse/FLINK-18958 Harold.Miao 于2020年8月13日周四 上午11:08写道： > hi all > 我发现TableColumn class不包含column comment ， 给开发带来了一点麻烦，请教大家一下，谢谢 > > > -- > > Best Regards, > Harold Miao >

Re: Re:Re:Re:Re:Flink SQL No Watermark

hi, watermark本来就是通过watermark assigner生成的。这是正常现象。 我想问问 你有没有试过调大并行度来解决这个问题？因为不同partition的数据可能存在时间上的差异。 Zhou Zach 于2020年8月13日周四 下午4:33写道： > > > > Hi forideal, Shengkai Fang, > > 加上env.disableOperatorChaining()之后，发现5个算子， > > > > > Source: TableSourceScan(table=[[

Re: Re:Re:Flink SQL No Watermark

hi 那你有没有试过将并行度设置为partition的数量 Zhou Zach 于2020年8月13日 周四下午3:21写道： > > > > Hi forideal, > 我也遇到了No Watermark问题，我也设置了table.exec.source.idle-timeout 参数，如下： > > > val streamExecutionEnv = > StreamExecutionEnvironment.getExecutionEnvironment > >

Re: Flink SQL No Watermark

hi, 你的意思是没有办法在codegen出来的代码上加断点的意思吗？ 这里倒是有一个比较hack的方法： 将生成的类放在一个java文件之中，然后修改改下GeneratedClass下的newInstance方法，如果classname == “WatermarkGenerator$2” 则将刚才的类则返回 new WatermarkGenerator$2 这个类。 我个人对于问题的猜测是有一条数据的rowtime远远晚于其他数据，从而将整体的watermark提得很高，导致后面的“晚到”的数据一直无法触发watermark的生成。 forideal 于2020年8月13日周四

Re: 关于FlinkSQL的一些最佳实践咨询

针对（3）社区建议使用sql api， table api现在正准备重构。 靳亚洽 于2020年8月13日周四 上午11:00写道： > 针对2， 我们公司采用的是通过提交一个解析flink sql的jar包，把sql作为jar包的参数来执行的。采用接口机将用户的udf > jar包和平台提供的解析flink sql的jar包提交到集群。 > 针对3， 既然使用了flink sql,当然期望sql搞定一切了 > 针对4， > 我们改造了下flink-client模块的一小块代码，支持提交多jar包到集群，所以connector那些包就通过这种方式上传了。当然提前放集群上面也是可以的。

Re: FlinkSQL even time 定义问题

对于第一个问题- 在查询语句之中定义watermark： 现在并不支持。这主要是由于在同一个作业之中，如果select的数据源是同一个表，那么在实际的优化过程之中，会将source进行复用，而现在同一个source并不支持多个watermark assigner。如果在不同的作业之中，那么显然只要修改watermark的定义语句即可。 对于第二个问题：rowtime的定义是必须建立在创建表的过程之中的。 对于第三个问题：社区正在讨论这个问题。现在仅支持多个insert的sql在同一个job之中。 Zhao,Yi(SEC) 于2020年8月12日周三 下午5:36写道： >

Re: flink1.11 sql使用问题

能展示下完整的例子吗？ 我用的时候没有这种情况。 魏烽 于2020年8月11日 周二下午10:27写道： > 各位大佬好： > > 在使用flink1.11 sql客户端的时候，只能只用最基本的count，group by、order > by、join等都无法实现，请问这个是什么原因呢，感谢！ > > > Flink SQL> select count(t2.superid) from cdp_profile_union t1 inner join > cdp_crowd_10002 t2 on t1.superid=t2.superid; > > [ERROR] Could not

Re: 来自郭华威的邮件

hi. 能提供具体的代码？ 郭华威 于2020年8月10日周一 上午10:21写道： > flink1.11.1 使用tableApi 报错： > Exception in thread "main" org.apache.flink.table.api.TableException: > Create BatchTableEnvironment failed. > at > org.apache.flink.table.api.bridge.java.BatchTableEnvironment.create(BatchTableEnvironment.java:517) >

Re: 关于Flink1.11 CSV Format的一些疑问

hi, 对于第一个问题，文档[1]中已经有较为详细的解释，你可以仔细阅读下文档关于partition files的解释。 对于第二个问题，现在的csv格式的确不支持这个选项，可以考虑见个jira作为improvment. [1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/filesystem.html WeiXubin <18925434...@163.com> 于2020年8月8日周六 上午11:40写道： > Hi，我在Flink1.11版本，使用filesystem

Re: 使用flink-sql-connector-elasticsearch6_2.11_1.10.0.jar执行任务失败

不好意思，在es6上也进行了相应的修复。 但似乎是一个相同的问题。 Shengkai Fang 于2020年8月7日周五 下午7:52写道： > 你的意思是不是用1.10的es包没问题，但是用1.11的有问题？ > 似乎是一个bug，在1.11 es7上已经修复了这个bug，但是没有对于es6进行修复。 > 参见[1] https://issues.apache.org/jira/browse/FLINK-18006?filter=-2 > > 费文杰 于2020年8月7日周五 下午3:56写道： > >&g

Re: 使用flink-sql-connector-elasticsearch6_2.11_1.10.0.jar执行任务失败

你的意思是不是用1.10的es包没问题，但是用1.11的有问题？ 似乎是一个bug，在1.11 es7上已经修复了这个bug，但是没有对于es6进行修复。 参见[1] https://issues.apache.org/jira/browse/FLINK-18006?filter=-2 费文杰 于2020年8月7日周五 下午3:56写道： > > 以下是我的代码： > import com.alibaba.fastjson.JSONObject; > import lombok.extern.slf4j.Slf4j; > import

  1   2   >