Re:Re: flink1.11编译失败

Hi, 这个错误应该是找到了老的这个类的 .class 文件。 你可以编译下整个工程"mvn clean install -DskipTests"。 Best, Hailong 在 2020-11-27 14:20:42，"zhy" 写道： >hi、 >感谢，确实是这个问题，我现在跑测试用例遇到下面问题，请问什么原因哪，我看文件是存在的 >Error:(32, 25) object contrib is not a member of package org.apache.flink >import

Re: flink1.11编译失败

hi、 感谢，确实是这个问题，我现在跑测试用例遇到下面问题，请问什么原因哪，我看文件是存在的 Error:(32, 25) object contrib is not a member of package org.apache.flink import org.apache.flink.contrib.streaming.state.RocksDBStateBackend Leonard Xu 于2020年11月26日周四 下午7:58写道： > HI > > 这两个类是 codegen 生成的，所以源码里没有，你编译下flink-sql-parser模块就会自动生成这几个类。 >

Re:Re: canal-json 分库分表场景应用

是的 感觉国内canal用户应该比debezium用户多很多，谢谢支持 在 2020-11-27 11:40:55，"Leonard Xu" 写道： >Hi, air23 > >国内用户用Canal-json还是很多的，我建了个issue 来支持, 你可以关注下。 > >> 你好 请问这个debezium-json 这个value.source.table 功能 在1.12的canal-json会实现吗， >> 看到canal-json代码里面 好像是有这部分代码。 > > >Best, >Leonard

Re: canal-json 分库分表场景应用

Hi, air23 国内用户用Canal-json还是很多的，我建了个issue 来支持, 你可以关注下。 > 你好 请问这个debezium-json 这个value.source.table 功能 在1.12的canal-json会实现吗， > 看到canal-json代码里面 好像是有这部分代码。 Best, Leonard [1]https://issues.apache.org/jira/browse/FLINK-20385 > > > > > >

Re: flink on native k8s 没有

你的提交命令是没有问题的，需要确认一下Flink client和JM的rest endpoint的联通情况 可以curl一下flink run命令打出来的JM rest地址，看看网络通不通 Best, Yang jiangjiguang719 于2020年11月25日周三 下午8:28写道： > flink版本：1.11 > k8s版本：1.18 > flink on native k8s 按照官网配置，可以看到jobmanager ui > 但是没有tm为0 slot为0 提交任务 就会一直卡主 > > > 角色配置： > [root@node20 rbac]# cat

Re: flink sql cdc 如果只处理一次全量数据问题

你说的有变化是后续的数据库进行增删改操作吗，如果是的话你从checkpoint启动就好了啊 cljb...@163.com 于2020年11月27日周五 上午11:10写道： > 之前一直使用streaming api，这两天开始使用sql。 > 有个疑问，flink sql cdc读取mysql的数据时候，会处理 全量 + 增量数据。 > 那么如果同一个任务上线后，后续有变化，修改后再次上线，这个时候我并不希望处理之前过的数据。这个时候是怎么做呢？ > > >

flink sql cdc 如果只处理一次全量数据问题

之前一直使用streaming api，这两天开始使用sql。 有个疑问，flink sql cdc读取mysql的数据时候，会处理 全量 + 增量数据。 那么如果同一个任务上线后，后续有变化，修改后再次上线，这个时候我并不希望处理之前过的数据。这个时候是怎么做呢？ cdc里面有进行state保存消费过的changelog的位置吗？这样我重新上线的时候从savepoint或者checkpoint进行恢复，是不是就可以了？ 感谢！ cljb...@163.com

Re: Re: canal-json 分库分表场景应用

1.12 马上要发布了，canal-json 中没来得及实现。 On Fri, 27 Nov 2020 at 10:49, air23 wrote: > 你好 请问这个debezium-json 这个value.source.table 功能 在1.12的canal-json会实现吗， > 看到canal-json代码里面 好像是有这部分代码。 > > > > > > > > > > > > > > > > > > 在 2020-11-27 10:13:37，"Jark Wu" 写道： > >目前还不支持，读取 table 元信息。 > >在 1.12

Re:Re: canal-json 分库分表场景应用

你好 请问这个debezium-json 这个value.source.table 功能 在1.12的canal-json会实现吗， 看到canal-json代码里面 好像是有这部分代码。 在 2020-11-27 10:13:37，"Jark Wu" 写道： >目前还不支持，读取 table 元信息。 >在 1.12 中，debezium-json 支持了这种功能，文档[1], 代码[2]。 >canal-json的话需要按照类似的方式支持下元信息读取。 > >Best, >Jark > >[1]:

Re: canal-json 分库分表场景应用

目前还不支持，读取 table 元信息。 在 1.12 中，debezium-json 支持了这种功能，文档[1], 代码[2]。 canal-json的话需要按照类似的方式支持下元信息读取。 Best, Jark [1]: https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/formats/debezium.html#available-metadata [2]:

canal-json 分库分表场景应用

你好 我这边有很多这种场景，把分库分表的binlog 写入到同一个topic 然后用canal-json解析，这边想获取到json里面的table 字段， 然后根据 table名称加主键 写入到下游 合成一张表，写入到下游表， 然后发现 canal-json 是获取不到表名的，然后这边去修改canal-json的format。 添加 createJsonRowType方法的DataTypes.FIELD("table", DataTypes.STRING()) 然后在deserialize方法里面把 table字段set到 data里面去。但是发现这种好像是不成功的 ，请问下

FlinkSQL使用Hive自定义函数报类找不到

Hi，FlinkSQL使用Hive的自定义函数，是不是需要将自定义的UDF包放入到flink/lib下 HiveCatalog和HiveModule都是加载了的，使用listFunctions是可以打印出自定义函数的 具体报错是在CatalogFunctionImpl的isGeneric方法中

Re:flink sql实时计算分位数如何实现

Hi, `Percentile` 函数应该是 hive 内置的 UDF，在 Flink SQL 可以直接使用 Hive 的内置 UDF， 具体使用方式可查阅文档[1]. [1]https://ci.apache.org/projects/flink/flink-docs-master/dev/table/connectors/hive/hive_functions.html Best, Hailong 在 2020-11-26 21:55:35，"爱成绕指柔" <1194803...@qq.com> 写道： >你好： > 目前flink

flink sql??????????????????????

?? flink sql??percentile

Re: 发现flinksql写hive比写hdfs慢很多

Hi Hive 使用的 BulkWriter 目前有两个问题，一个是支持的数据类型没有MR writer的全，第二个是，BulkWriter 支持parquet和orc，但只支持orc的最新版本，写入低版本有兼容性问题 主要因为这两个问题，所以默认使用MR writer。 祝好， Leonard > 在 2020年11月26日，20:05，admin <17626017...@163.com> 写道： > > BulkWriter

Re: FlinkSQL导致Prometheus内存暴涨

IIRC, runtime will truncate the operator name to max 80 characters, see `TaskMetricGroup#METRICS_OPERATOR_NAME_MAX_LENGTH`. You can search the log if there are "The operator name {} exceeded the {} characters length limit and was truncated.". On Thu, 26 Nov 2020 at 18:18, hailongwang

Re:Re:udtf collect方法 报NEP

Hi, 1、NPE问题，https://github.com/apache/flink/blob/7a490d85ab4113f859e5ca8e2cd163439452c221/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/collector/WrappingCollector.java#L43 在这里collect的时候DEBUG发现this.collector为null, https://imgchr.com/i/D0Hm5Q 这里我打了断点； “第一次调用正常”

Re: 发现flinksql写hive比写hdfs慢很多

既然性能差异这么大，所以为什么不默认使用BulkWriter，而使用MR writer呢 > 2020年11月26日 下午7:50，Leonard Xu 写道： > > > Hi, admin > > 谢谢验证， >> 在 2020年11月26日，17:43，admin <17626017...@163.com> 写道： >> >> 默认true的情况下 两个任务同时写30分钟，写hive的任务就已经落后了3分钟 > > 此时，写hive用MR writer，写HDFS只支持Flink BulkWriter, 如果单独测试sink的话，Flink BulkWriter >

Re: flink1.11编译失败

HI 这两个类是 codegen 生成的，所以源码里没有，你编译下flink-sql-parser模块就会自动生成这几个类。 祝好， Leonard > 在 2020年11月26日，19:43，zhy 写道： > > hi、flink1.11 release source编译为什么会缺失类文件，去github仓库也没找到，如何解决这个问题～ > > > import org.apache.flink.sql.parser.impl.ParseException; > > import

Re: 发现flinksql写hive比写hdfs慢很多

Hi, admin 谢谢验证， > 在 2020年11月26日，17:43，admin <17626017...@163.com> 写道： > > 默认true的情况下 两个任务同时写30分钟，写hive的任务就已经落后了3分钟 此时，写hive用MR writer，写HDFS只支持Flink BulkWriter, 如果单独测试sink的话，Flink BulkWriter 应该不止10%的性能提升。 > false的情况，两个写30多分钟，差异不大 false时，两个作业都用 Flink BulkWriter，差异是不大的。 另外

flink1.11编译失败

hi、flink1.11 release source编译为什么会缺失类文件，去github仓库也没找到，如何解决这个问题～ import org.apache.flink.sql.parser.impl.ParseException; import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl; Error:(39, 87) java: 找不到符号 符号: 类 ParseException 位置: 类

Flink WebUI上每个TaskManager显示的cpu和mem是怎么检测到的。

如题，有具体的检测命令嘛。 我用的公司的pass环境，今天发现有个检测到core为1的（但可以确定该机器实际core很多）。

Re:udtf collect方法 报NEP

Hi， +I 表示是一条 insert 的数据，其它类型的可以查看： https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/types/RowKind.java 其中 NPE 问题，你是用哪个版本的呢？ 方便复制下 sql 和 udtf 吗，我看下能不能复现。 PS：“第一次调用正常” 是指？ Best， Hailong 在 2020-11-26 17:25:03，"bulterman" <15618338...@163.com> 写道：

udtf collect方法 报NEP

Hi all, 我在使用udtf 调用collect方法的时候出现空指针异常，显示collector对象为null,第一次调用正常，第二次调用数据被标记了 “INSERT”是什么意思？ https://imgchr.com/i/D0HuCj https://imgchr.com/i/D0Hm5Q udtf代码： https://imgchr.com/i/D0HeUg

Re:FlinkSQL导致Prometheus内存暴涨

Hi， 是的，个人觉得可以提供一个配置项来控制 task Name。 完整的 task name 有助于排查问题等，简短的 task name 有助于在生产环境中 metric 的采集，可以极大较少发送的网络开销，存储空间等。 已建立个了 issue ：https://issues.apache.org/jira/browse/FLINK-20375 Best, Hailong 在 2020-11-24 14:19:40，"Luna Wong" 写道： >FlinkSQL 生成的Metrics数据

Re: 发现flinksql写hive比写hdfs慢很多

Hi,Leonard 我将这个issue的改动合到我的代码里，同时看到有邮件列表里提到 table.exec.hive.fallback-mapred-writer 这个配置对写入速度也有影响， 所以我分别基于true或false做了测试。 结果是： 默认true的情况下 两个任务同时写30分钟，写hive的任务就已经落后了3分钟 false的情况，两个写30多分钟，差异不大 所以使用MR writer和flink native writer在性能上确实有很大差异 > 2020年11月26日 下午5:32，Leonard Xu 写道： > > Hi, admin > 结合这个

Re: 发现flinksql写hive比写hdfs慢很多

Hi, admin 结合这个 issue 和你的对比结果， 我觉得应该是这个bug，这个问题在最新的分支已经修复，今天社区cut branch了，你可以帮忙在1.12的分支或master的分支上验证下吗？ 祝好， Leonard [1] https://github.com/apache/flink/tree/release-1.12 [2] https://github.com/apache/flink/tree/master

Re: flink1.11.2 一直不入hbase问题

HI 图挂了，Flink 1.11.x 版本hbase的 identifier 都是 hbase-1.4, 不是 hbase-1.4.3, 应该会报错才对，可以确认下日志里也没有报错吗？ 祝好 Leonard > 在 2020年11月26日，14:05，simon 写道： > > hbase-1.4.3

Re: flink 1.11.2 cdc: cdc sql 结果表的sink顺序问题， 不同并行度下从save point中恢复job时，会导致sink结果不对！！

Btw, I created an issue to track this problem: https://issues.apache.org/jira/browse/FLINK-20374 Hope we can fix it in the next versions to have a better out-of-box experience. Best, Jark On Thu, 19 Nov 2020 at 13:58, Jark Wu wrote: > 如果数据本身没什么倾斜，且并发也能打上去。那在 sql 这边也没什么其他办法了。得从 rocksdb

flinksql????format??avro??kafka topic ???? ArrayIndexOutOfBoundsException

flinksqlCREATE TABLE kafkaInputTable ( name String, address String ) WITH ( 'connector' = 'kafka', 'topic' = 'test', 'properties.bootstrap.servers' = 'node1:9092,node2:9092,node3:9092', 'format' = 'avro' ); CREATE TABLE kafkaOutputTable ( name String, address String ) WITH (

flinksql????format??avro??kafka topic ???? ArrayIndexOutOfBoundsException

flinksqlCREATE TABLE kafkaInputTable ( name String, address String ) WITH ( 'connector' = 'kafka', 'topic' = 'test', 'properties.bootstrap.servers' = 'node1:9092,node2:9092,node3:9092', 'format' = 'avro' ); CREATE TABLE kafkaOutputTable ( name String, address String ) WITH (