Re: flink-kafka-sink

hi，没有效果 具体是啥？ cxx <1156531...@qq.com> 于2021年1月7日周四 上午9:53写道： > 我从kafka消费一条数据，然后将消息进行切分，再发送到下游的kafka中，但是这样不能保证在一个事务里面。 > 例如：我将一条数据切分成10条，然后再第五条的时候抛出一个异常，但是前四条已经发送到下游的kafka了。 > 我设置了事务id，隔离级别，client > id，enable.idempotence,max.in.flight.requests.per.connection,retries > 但是没有效果。 > > > > -- > Sen

flink watermark 为负数的异常

hi，我通过webUI查看了当前的程序，watermark变为-2，一直没变过，出现这种情况是什么原因，一头雾水。截图在附件里，帮忙分析一下。

Re: flink sql读kafka元数据问题

hi 你写入数据的时候设置 headers 了吗 没设置的话当然是空的了 - Best Wishes JasonLee -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Re: idea导入flink源码问题(看不到图片，贴了下文字)

建议你把Maven的settings.xml以及你项目的pom.xml也贴出来看看，大家才好帮你定位问题 penguin. 于2021年1月13日周三 下午2:25写道： > 贴不了图，我直接放文字吧 > > > > ▼θSync: at 2021/1/13 12:05 with 18 errors > >▼Resolve dependencies 4 errors > > Cannot resolve netminidev:json-smart:2.3 > > Cannot resolve io.confluent:kafka-schema

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

Hi 唐老师， 谢谢解惑，明白了。 祝好 Yun Tang 于2021年1月13日周三 下午2:59写道： > Hi > > 其实问题本质很简单，作为一个分布式计算引擎，需要一个分布式文件系统才能保证可以做到分布式检查点的创建和恢复 > [1]。没有必要再尝试将checkpoint写到本地磁盘上，否则一旦发生failover是无法正常恢复的。 > > > [1] > https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/checkpointing.html#prerequisites > >

Flink 1.12 Too old resource version

Flink 版本： flink1.12.0 K8s版本： 1.17 Docker 镜像: apache/flink:1.12.0-scala_2.11 使用以上的docker 镜像在k8s中 session模式下还是会报： "Too old resource version" exception in Kubernetes watch more gracefully Jira上也有对应的issue, 对于1.12.0 版本如何修复呢？

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

Hi 其实问题本质很简单，作为一个分布式计算引擎，需要一个分布式文件系统才能保证可以做到分布式检查点的创建和恢复 [1]。没有必要再尝试将checkpoint写到本地磁盘上，否则一旦发生failover是无法正常恢复的。 [1] https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/checkpointing.html#prerequisites 祝好 唐云 From: Carmen Free Sent: Wednesday

Flink sql cli 查询hive 为两个字段的分区表如果where条件不指定这两个字段条件会出异常

生产的hive表由datekey和event两个字段作为分区，查询sql语句： （1）第一组对比 SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(*正常*) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (*异常*) （2）第二组对比 SELE

Flink sql cli 查询hive 为两个字段的分区表如果where条件不指定这两个字段条件会出异常

生产的hive表由datekey和event两个字段作为分区，查询sql语句： （1）第一组对比 SELECT vid From table_A WHERE datekey = '20210112' AND event = 'XXX' AND vid = 'aa';(*正常*) SELECT vid From table_A WHERE datekey = '20210112' AND vid = 'aa'; (*异常*) SELECT vid From ta

Re: 为什么我这个打印的是上次的watermark,而是用的是这次的watermark呢,为什么打印的不是这次的呢

描述还是不清晰。 watermark是定期生成，你获取的时候不一定已经更新。 何宗谨 于2021年1月13日周三 上午10:20写道： > > 允许的时间间隔是3秒,每次打印的都是上一个时间戳的watermark,但是使用的好像还是这次的 > > -- 原始邮件 -- > *发件人:* "user-zh" ; > *发送时间:* 2021年1月13日(星期三) 上午10:02 > *收件人:* "user-zh"; > *主题:* Re: 为什么我这个打印的是上次的watermark,而是用的是这次的watermark呢,

Re:Re: idea导入flink源码问题(看不到图片，贴了下文字)

贴不了图，我直接放文字吧 ▼θSync: at 2021/1/13 12:05 with 18 errors ▼Resolve dependencies 4 errors Cannot resolve netminidev:json-smart:2.3 Cannot resolve io.confluent:kafka-schema-registry-client:4.1.0 Cannot resolve com.nimbusds:nimbus-jose-jwt:9.4.1 Cannot resol

Re:Re: idea导入flink源码问题

好像图放在正文里面有问题，放在附件里了 在 2021-01-13 13:42:27，"Carmen Free" 写道： >hi， > >图看不见。 > >penguin. 于2021年1月13日周三 下午1:19写道： > >> 已经在maven的setting文件中配置了好几个镜像了，还是这样。如下图 >> >> >> >> >> >>

Re:Re: idea导入flink源码问题

Hi, 图好像是有点问题，我重传一下 在 2021-01-13 13:42:27，"Carmen Free" 写道： >hi， > >图看不见。 > >penguin. 于2021年1月13日周三 下午1:19写道： > >> 已经在maven的setting文件中配置了好几个镜像了，还是这样。如下图 >> >> >> >> >> >>

Re: idea导入flink源码问题

hi， 图看不见。 penguin. 于2021年1月13日周三 下午1:19写道： > 已经在maven的setting文件中配置了好几个镜像了，还是这样。如下图 > > > > > >

如何修改FlinkSQL中DATE类型对应的Conversion类？

Hi: Flink SQL 中，每种数据类型都有默认的Conversion类，有些数据类型支持多种，比如DATE，支持java.sql.Date、java.time.LocalDate、Integer。而DATE默认的是java.time.LocalDate，我现在希望在将其更改为java.sql.Date，可以通过什么设置将其更改？ 祝好！ automths

idea导入flink源码问题

已经在maven的setting文件中配置了好几个镜像了，还是这样。如下图

flink源码项目导入idea，无法下载插件和依赖

如图，在网上查找配了好几个maven镜像了，求助！！！

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

Hi 唐老师， 我又重新尝试了新的场景。 我将集群起来后，然后使A节点的TM失效（ 此时集群中仅有A节点的JM, B节点的TM ），这个时候在flnk web ui界面提交新的任务，任务被调度到B节点TM，可以发现任务一直正常运行，但是trigger检查点快照时一直不成功。报错跟前文中描述恢复检查点时的错误很类似，只不过恢复检查点时，是找不到chk-xx文件，在这里是无法创建chk-xx文件。 具体报错如下： 2021-01-13 10:33:10,042 WARN org.apache.flink.runtime.checkpoint.CheckpointCoordinator

flink sql读kafka元数据问题

WITH("`event_time` TIMESTAMP(3) METADATA FROM 'timestamp'," + "`partition` BIGINT METADATA VIRTUAL," + "`offset` BIGINT METADATA VIRTUAL," + "`headers` MAP NOT NULL METADATA VIRTUAL,”+ 在这里获取kakfa元数据时，官网没有说怎么获取kafka消息的key？，headers的信息是空的，请问怎么在flink sql中获取kafka消息key？

Re: Re: flink waterMark 相关问题

private long autoWatermarkInterval = 200; /** * Interval in milliseconds for sending latency tracking marks from the sources to the sinks. */ 张锴 于2021年1月13日周三 上午10:26写道： > ok，明白了 > > anonnius 于2021年1月13日周三 上午10:20写道： > > > 在 StreamExecutionEnvironmennt的方法@PublicEvolving public > > voi

Re: Re: flink waterMark 相关问题

ok，明白了 anonnius 于2021年1月13日周三 上午10:20写道： > 在 StreamExecutionEnvironmennt的方法@PublicEvolving public > void setStreamTimeCharacteristic(TimeCharacteristic characteristic) { > this.timeCharacteristic = > Preconditions.checkNotNull(characteristic); if > (characteristic

?????? ??????????????????????????watermark,????????????????watermark??,????????????????????????

3??,watermark, --  -- ??: "user-zh"

?????? flink waterMark ????????

master??200??1.10watermark --  -- ??: "user-zh"

?????? flink waterMark ????????

master??200??1.10watermark --  -- ??: ""

Re:Re: flink waterMark 相关问题

在 StreamExecutionEnvironmennt的方法@PublicEvolving public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) { this.timeCharacteristic = Preconditions.checkNotNull(characteristic); if (characteristic == TimeCharacteristic.ProcessingTime) {

Re: flink waterMark 相关问题

我从ExecutionConfig找到了，private long autoWatermarkInterval = 0，并不是200毫秒，这个代表一个时间戳就代表一个watermark是吗 Ball's Holy <873925...@qq.com> 于2021年1月13日周三 上午9:42写道： > hi 张锴, > 如果我没记错的话，默认时间间隔是200毫秒，具体设置细节可以看flink.api.common.ExecutionConfig > 对应的属性autoWatermarkInterval > > > > > -- 原始邮件 -

Re: 为什么我这个打印的是上次的watermark,而是用的是这次的watermark呢,为什么打印的不是这次的呢

图挂了。 何宗谨 于2021年1月13日周三 上午9:20写道： > > > > >

Re: flink waterMark 相关问题

ok，谢谢你 Ball's Holy <873925...@qq.com> 于2021年1月13日周三 上午9:42写道： > hi 张锴, > 如果我没记错的话，默认时间间隔是200毫秒，具体设置细节可以看flink.api.common.ExecutionConfig > 对应的属性autoWatermarkInterval > > > > > -- 原始邮件 -- > 发件人: "anonnius" 发送时间: 2021年1月13日(星期三) 上午9:19 > 收件人: "user-zh" 主题: Re:flink w

Re: flink waterMark 相关问题

谢谢你 anonnius 于2021年1月13日周三 上午9:19写道： > 可以看一下 ExecutionConfig这个类 > 在 2021-01-12 17:55:47，"张锴" 写道： > >hi，我使用的flink版本是1.10，我想问一下watermark的默认时间间隔是多少，以及所对应的源码位置在哪里，我并没有找到。 >

flink1.11.1 如何让多个log4j配置文件生效

dear all： 我的flink任务提交到yarn运行， 默认生效的是日志配置是flink/conf中的log4j.properties。 但我的应用jar包中还有一个log4j2.xml，这里面配置了KafkaAppend，要把日志发送到kafka。 我要如果设置，才能让这两个配置文件都生效呢？ 哪位大侠有配置经验。 thanks by nicygan

Re: Flink SQL一个Job使用多个Catalog的例子

我已经解决了。USE其他Catalog再建表即可。 Luna Wong 于2021年1月12日周二 下午9:41写道： > > 大家好。 > 我没有在官网找到个Job使用多个Catalog的例子。 > 我想在一个Job里使用普通的Catalog注册个Kafka Source然后，将数据发送Iceberg Sink表。这个Sink表 > 注册在另一个Iceberg + Hive 的Catalog 中。 > 注册代码如下。 > CREATE CATALOG hive_catalog WITH ( > 'type'='iceberg', > 'catalog-type'='hive

??????flink waterMark ????????

hi , 200flink.api.common.ExecutionConfig ??autoWatermarkInterval --  -- ??: "anonnius"

??????????????????????????watermark,????????????????watermark??,????????????????????????

Re:flink waterMark 相关问题

可以看一下 ExecutionConfig这个类 在 2021-01-12 17:55:47，"张锴" 写道： >hi，我使用的flink版本是1.10，我想问一下watermark的默认时间间隔是多少，以及所对应的源码位置在哪里，我并没有找到。

Re: FlinkSQL Filter Error With Float Column on flink-1.12.0

你好，在数据库中，Float类型存的是个近似值，不能用类似于 = 或者 != 的比较语句，所以也不支持IN操作 希望能帮助到你 From: jy l Date: 2021-01-12 18:04 To: user-zh Subject: FlinkSQL Filter Error With Float Column on flink-1.12.0 Hi: Flink SQL filter data throw an exception, code: def main(args: Array[String]): Unit = { val env = StreamExecut

回复: flink 编译

你好： 你可以在idea中mvn package 或者mvn install 相应的模块 hdxg1101300...@163.com 发件人： penguin. 发送时间： 2021-01-12 21:44 收件人： user-zh@flink.apache.org 主题： flink 编译 Hi， 请问有人知道怎么单独编译flink-runtime模块吗？ 然后这样是否能把更改的部分直接在flink-dist包中的org.apache.flink.runtime目录下进行替换？ 整体编译一次实在太慢了。 谢谢！ penguin

Re: flink 编译

试试 mvn clean install -DskipTests -pl flink-runtime,flink-dist Best, tison. penguin. 于2021年1月12日周二 下午9:44写道： > Hi， > > > 请问有人知道怎么单独编译flink-runtime模块吗？ > 然后这样是否能把更改的部分直接在flink-dist包中的org.apache.flink.runtime目录下进行替换？ > 整体编译一次实在太慢了。 > 谢谢！ > > > penguin

flink 编译

Hi， 请问有人知道怎么单独编译flink-runtime模块吗？ 然后这样是否能把更改的部分直接在flink-dist包中的org.apache.flink.runtime目录下进行替换？ 整体编译一次实在太慢了。 谢谢！ penguin

Flink SQL一个Job使用多个Catalog的例子

大家好。 我没有在官网找到个Job使用多个Catalog的例子。 我想在一个Job里使用普通的Catalog注册个Kafka Source然后，将数据发送Iceberg Sink表。这个Sink表 注册在另一个Iceberg + Hive 的Catalog 中。 注册代码如下。 CREATE CATALOG hive_catalog WITH ( 'type'='iceberg', 'catalog-type'='hive', 'uri'='thrift://kudu1:9083', 'clients'='2', 'property-version'='1', 'warehous

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

Hi， 从异常日志看，应该是因为你的state.checkpoints.dir 或者说 statebackend的checkpoint目录配置成了本地目录，checkpoint保存到了本地机器上，所以在failover restore的时候，必须得让原task部署回原来的机器才能正常运行。将state backend的checkpoint目录更换为一个DFS目录即可。 祝好 唐云 From: Carmen Free Sent: Tuesday, January 12, 2021 18:14 To: user-zh@

Re: flink编译报错

设置下镜像可以。 Yun Tang 于2021年1月12日周二 下午5:37写道： > Hi, > > 国内网络环境不太好，其实问题是node.js 安装有问题，可以考虑单独安装一下node > js和npm，如果还是不行，在不需要webui的前提下，可以编译时候加上profile “-Pskip-webui-build” > 来skip掉该部分的编译。 > > 祝好 > 唐云 > > From: Ruguo Yu > Sent: Tuesday, January 12, 2021 14:00 > To: user-z

回复：flink版本升级问题咨询

描述的不太对，具体可以参考下这个 https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/libs/state_processor_api.html 在2021年1月11日 20:12，xuhaiLong 写道： 我试过 flink 1.7 升级到 1.10。如果使用到了 table api 涉及 group By 的话 savePoint恢复有问题，其他没发现过什么问题。或者可以使用下 process api 写一份数据，启动 job 在2021年1月7日 09:50，zhang hao 写道： 目前现状：公

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

你好，唐老师，谢谢解答。 不好意思，下面补充一下报错信息，刚才忘记说了。 主要报错信息如下,重新模拟了下： 2021-01-12 18:09:34,236 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph- Source: Custom Source -> Flat Map -> Timestamps/Watermarks (2/2) (3b50a7ce56b408c2978260846b76a28a) switched from RUNNING to FAILED on org.apache.fli

FlinkSQL Filter Error With Float Column on flink-1.12.0

Hi: Flink SQL filter data throw an exception, code: def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment val source = env.fromElements( (1.0f, 11.0f, 12.0f), (2.0f, 21.0f, 22.0f), (3.0f, 31.0f, 32.0f), (4.0f, 41.0f, 42.

flink waterMark 相关问题

hi，我使用的flink版本是1.10，我想问一下watermark的默认时间间隔是多少，以及所对应的源码位置在哪里，我并没有找到。

Re: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

Hi Flink的容错机制是可以保证TM lost时候会尝试重启作业，“为何任务不能恢复”是需要看完整异常栈的，简单描述是无法帮助排查问题的。 祝好 唐云 From: Carmen Free Sent: Tuesday, January 12, 2021 15:52 To: user-zh@flink.apache.org Subject: rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？ hi, rocksdb作为statebackend时，TM节点挂掉了，为何任务不能恢复呢？

Re: Flink代码一直报 java.lang.NullPointerException错误，但变量明明初始化了，却仍然有该错误，找不到原因，请诸位指教。谢谢

Hi， 这个错误其实是kryo初始化时候扔出来的。你自定义的类 SockRowV2，WashDetectionSockValue 等，不符合Flink关于pojo的定义，所以会回退到使用kryo进行序列化/反序列化。建议将相关类在kryo上进行注册 [1]。特别地，如果是thrift或者protobuf的类，需要单独注册[2]，更好的方法其实是建议将你们的自定义类修改为满足Flink的POJO类 [3] [1] https://flink.apache.org/news/2020/04/15/flink-serialization-tuning-vol-1.html#kryo [

Re: flink编译报错

Hi, 国内网络环境不太好，其实问题是node.js 安装有问题，可以考虑单独安装一下node js和npm，如果还是不行，在不需要webui的前提下，可以编译时候加上profile “-Pskip-webui-build” 来skip掉该部分的编译。 祝好 唐云 From: Ruguo Yu Sent: Tuesday, January 12, 2021 14:00 To: user-zh@flink.apache.org Subject: Re: flink编译报错 试下这个命令 mvn clean insta

Re: 回复：Flink1.12.0 sql-client连接hive报错

1、现在sql-cli能够提交到yarn的session那边，但是会直接报错： Caused by: java.lang.ClassNotFoundException: org.apache.hadoop.mapred.JobConf at java.net.URLClassLoader.findClass(URLClassLoader.java:381) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at sun.misc.Launcher$AppClassLoader.lo

Re: Flink SQL 可以使用异步IO特性吗

hello,kenyore. 我大致了解了你的意思，你可以通过继承AsyncTableFunction的方式实现数据库异步IO。 公共抽象类AsyncTableFunction 扩展了UserDefinedFunction AsyncTableFunction

Re: Flink SQL 可以使用异步IO特性吗

hello,kenyore. 我大致了解了你的意思，你可以通过继承AsyncTableFunction的方式实现数据库异步IO。 公共抽象类AsyncTableFunction 扩展了UserDefinedFunction AsyncTableFunction