flink1.11 set yarn slots failed

Hi all, 使用如下命令，设置Number of slots per TaskManager /opt/flink-1.11.0/bin/flink run-application -t yarn-application \ -Djobmanager.memory.process.size=1024m \ -Dtaskmanager.memory.process.size=2048m \ -ys 4 \ 发现并不能override/opt/flink-1.11.0/bin/flink/conf/flink-conf.yaml中的默认值，每次要调整只能通过更改flink-conf

Re: 【求助】Flink Hadoop依赖问题

你可以在Pod里面确认一下/data目录是否正常挂载，另外需要在Pod里ps看一下 起的JVM进程里的classpath是什么，有没有包括hadoop的jar 当然，使用Roc Marshal建议的增加flink-shaded-hadoop并且放到$FLINK_HOME/lib下也可以解决问题 Best, Yang Roc Marshal 于2020年7月15日周三 下午5:09写道： > > > > 你好，Z-Z, > > 可以尝试在 > https://repo1.maven.org/maven2/org/apache/flink/flink-shaded-hadoop-2-

?????? ??????????: flink state

??thanku all --  -- ??: "Congxian Qiu"https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/stream/state/broadcast_state.html > [2] https://cloud.tencent.com/developer/article/1509789 > Best

Trying to eagerly schedule a task whose inputs are not ready

flink org.apache.flink.runtime.executiongraph.ExecutionGraphException: Trying to eagerly schedule a task whose inputs are not ready (result type: PIPELINED_BOUNDED, partition consumable: false, producer state: SCHEDULED, producer slot: null).at org.apache.flink.runti

Re: [sql-client] 通过sql-client提交sql怎么设置checkpointing.interval

是在flink-conf.yaml里面配置这个参数吗 execution.checkpointing.interval godfrey he 于2020年7月16日周四 下午1:37写道： > 现在还不支持在sql-client-defaults.yaml 里配置 checkpointing.interval, > 你可以配置在flink-conf.yaml里 >

Re: [sql-client] 通过sql-client提交sql怎么设置checkpointing.interval

现在还不支持在sql-client-defaults.yaml 里配置 checkpointing.interval, 你可以配置在flink-conf.yaml里 Harold.Miao 于2020年7月16日周四 下午1:27写道： > hi flink users > > 通过sql-client提交sql怎么设置checkpointing.interval？ > 我看了一下sql-client-defaults.yaml中的execution， 并没有发现这个参数。请教大家一下。 > 谢谢 > > > > -- > > Best Regards, > Harold Miao >

[sql-client] 通过sql-client提交sql怎么设置checkpointing.interval

hi flink users 通过sql-client提交sql怎么设置checkpointing.interval？ 我看了一下sql-client-defaults.yaml中的execution， 并没有发现这个参数。请教大家一下。 谢谢 -- Best Regards, Harold Miao

Re: Flink 1.11 submit job timed out

Hi 如果没有异常，GC 情况也正常的话，或许可以看一下 pod 的相关日志，如果开启了 HA 也可以看一下 zk 的日志。之前遇到过一次在 Yarn 环境中类似的现象是由于其他原因导致的，通过看 NM 日志以及 zk 日志发现的原因。 Best, Congxian SmileSmile 于2020年7月15日周三 下午5:20写道： > Hi Roc > > 该现象在1.10.1版本没有，在1.11版本才出现。请问这个该如何查比较合适 > > > > | | > a511955993 > | > | > 邮箱：a511955...@163.com > | > > 签名由 网

Re: 回复：答复: flink state

Hi broadcast state 是无法修改的，如果你还希望进行修改的话，可以使用 zhao liang 的方法，另外如果这个全局 state 不需要维护一致性等的话，同样可以考虑放到外存中（Redis，HBase 等） Best, Congxian zhao liang 于2020年7月15日周三 下午6:05写道： > Broadcast > state是无法满足你的要求的，估计你只能像我这样把涉及的state数据融入到数据流中，在算子中针对不同的类型数据做区分了，等于人工维持这个broadcast的流的变化。 > > 发件人: Robert.Zhang <1736

Re:Re: flink sql 1.11 create hive table error

hi Leonard, 感谢答疑！ At 2020-07-16 12:16:40, "Leonard Xu" wrote: >Hello, Zach > >是的, 1.12 会支持，PR[1]已经开了，在review中。 > >祝好， >Leonard Xu >[1] https://issues.apache.org/jira/browse/FLINK-18588 > >> 在 2020年7月16日，12:07，Zhou Zach 写道： >> >

Re: flink sql 1.11 create hive table error

Hello, Zach 是的, 1.12 会支持，PR[1]已经开了，在review中。 祝好， Leonard Xu [1] https://issues.apache.org/jira/browse/FLINK-18588 > 在 2020年7月16日，12:07，Zhou Zach 写道： > > Hi all, > flink sql 1.11 create table 是不是 不支持 IF NOT EXISTS > > > Query： >val hiveC

Re: flink 1.11 upsert结果出错

> 在 2020年7月16日，11:44，小学生 <201782...@qq.com> 写道： > > t_env.execute_sql('''delete from source_tab where trck_id='aew'  ''') 你这张表定义的是 Flink 中的表，这张表对应的是你外部系统(MySQL数据库)中的表，Flink 不支持 表上 的DELETE [1]， Flink 是一个计算引擎, 主要场景是读取、写入外部系统，修改外部系统的数据目前只发生在写入（insert）的时候，并且主要是为了保证数据一致性语义，需要往下游系统发Delete消息， 这个delet

Re: Flink-1.11内置connector测试问题求解

目前 1.11 版本中的 tableResult.print 只支持 exactly once 语义，需要配置 checkpoint。 1.12 里准备支持 at least once 语义，用户可以不用配置 checkpoint。目前 pr [1] 正在reivew 。 [1] https://github.com/apache/flink/pull/12867 Best, Godfrey Jingsong Li 于2020年7月16日周四 上午11:36写道： > tableResult.print需要有checkpoint > > Best, > Jingsong > >

flink sql 1.11 create hive table error

Hi all, flink sql 1.11 create table 是不是 不支持 IF NOT EXISTS Query： val hiveConfDir = "/etc/hive/conf" val hiveVersion = "2.1.1" val odsCatalog = "odsCatalog" val odsHiveCatalog = new HiveCatalog(odsCatalog, "ods", hiveConfDir, hiveVersion) streamTableEnv.registerCatalog(odsCa

Re: flink 1.11 upsert????????

??delete?? from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic, CheckpointingMode from pyflink.table import StreamTableEnvironment, EnvironmentSettings source=""" CREATE TABLE source_tab (  trck_id VARCHAR,  score  I

FlinkKafkaConsumer API 维表关联

各位好： 请教一下，用FlinkKafkaConsumer API的话，如何支持SQL的方式，和维表关联。（之前用Kafka API&StreamTableDescriptor API是可以的） "select a.id,b.name from kafka_table a " + "join dim_table FOR SYSTEM_TIME AS OF a.proctime as b on a.id = b.user_id"; thanks & Regards

Re: Flink-1.11内置connector测试问题求解

tableResult.print需要有checkpoint Best, Jingsong On Thu, Jul 16, 2020 at 11:31 AM amen...@163.com wrote: > hi, everyone > > 小白在测试flink > 1.11新特性新内置的三个connector时，在本地创建图片[1]中的任务并进行数据打印时，控制台只打印了表schema，而没有按内置的datagen > connector规则产生数据，请问可能是什么原因呢？谢谢解答！ > > > [1] https://postimg.cc/PprT9XV6 > > best,

Flink-1.11内置connector测试问题求解

hi, everyone 小白在测试flink 1.11新特性新内置的三个connector时，在本地创建图片[1]中的任务并进行数据打印时，控制台只打印了表schema，而没有按内置的datagen connector规则产生数据，请问可能是什么原因呢？谢谢解答！ [1] https://postimg.cc/PprT9XV6 best, amenhub amen...@163.com

Re: flink 1.11 upsert结果出错

Hi, 你需要什么样的例子，如果你用的table/sql的话，在官方文档对应的地方都有java/scala/python的对应写法。如果是python udf相关的东西，你可以参考[1] [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/python/ Best, Xingbo 小学生 <201782...@qq.com> 于2020年7月16日周四 上午11:14写道： > 谢谢两位大佬的解答，但是理解有点抽象，不太清楚，有没有pyflink下一个简单例子呢。

Re: flink 1.11 upsert????????

??pyflink??

Re: flink 1.11 upsert结果出错

Hi, Leonard 说的是对的，除了udf的部分，pyflink的所有的api都是调用的java端的功能，如果java端没有，pyflink就不支持 Best, Xingbo Leonard Xu 于2020年7月16日周四 上午11:09写道： > Hi， > > 我理解 pyflink 底层也会走到你看到的java代码, 我对 pyflink 不是很熟， cc xingbo 补充下。 > > 祝好 > Leonard Xu > > > 在 2020年7月16日，11:04，小学生 <201782...@qq.com> 写道： > > > > 各位大佬好，由于不是特别懂java

Re: flink 1.11 upsert结果出错

Hi， 我理解 pyflink 底层也会走到你看到的java代码, 我对 pyflink 不是很熟， cc xingbo 补充下。 祝好 Leonard Xu > 在 2020年7月16日，11:04，小学生 <201782...@qq.com> 写道： > > 各位大佬好，由于不是特别懂java,所以麻烦问下pyflink里面有相关mysql的delete吗，官网没看到，谢谢！

Re: flink 1.11 upsert????????

??java,pyflink??mysql??delete??

HELP，flink1.10 sql整合hbase，insert into时，提示validateSchemaAndApplyImplicitCast报错

Hi, 我在使用flink1.10.1的sql功能，hbase的版本是1.4.3，写入hbase时，提示validateSchemaAndApplyImplicitCast报错，意思是Query的Schema和Sink的Schema不一致。主要是Query Schema中的Row(EXPR$0)，里面都是表达式。Sink Schema中是Row(device_id)这种。我不知道，在sql中如何写，才能和hbase的sink schema保持一致。 我尝试了，类似于在select device_id as rowkey, ROW( device_id as 这里不能as ) as f1，

Re:Re: flink 1.11 sql类型问题

hi, leonard 感谢回复。我在es的ddl with参数里加了这个，貌似还是报错。我再简单描述下我的场景： 我的es sink的ddl如下： create table es_sink ( a varchar, b varchar, c TIMESTAMP(9) WITH LOCAL TIME ZONE ) with ( ) 我使用处理时间属性，将流里的proctime转成UTC格式的日期类型，作为c这个字段写入。现在能原生支持么？之前在1.10版本貌似是可以直接写的。但是到1.11写的不带时区了，导致不能兼容之前的格式。

Re: flink 1.11 sql类型问题

Hello json解析UTC时间是支持的，你with参数里指定下json中timestamp的类型试下， json.timestamp-format.standard = 'ISO-8601' Best Leonard Xu [1] https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/formats/json.html#json-timestamp-format-standard

Re:flink 1.11 sql类型问题

hi, 我通过flink sql 定义了一个es sink，其中有个字段类型定义为了 eventTime TIMESTAMP(9) WITH LOCAL TIME ZONE。 在尝试写入时，报了如下的异常。看来json parser无法解析这种类型。请问下大神们，我应该怎么写入一个UTC日期的时间类型？格式类似 2020-07-15T12:00:00.000Z java.lang.UnsupportedOperationException: Not support to parse type: TIMESTAMP(9) WITH LOCAL TIME ZONE at org

Re:Re: flink 1.9.2 升级 1.10.0 任务失败不能从checkpoint恢复

Hello， Peihui，可以参考下是不是和这个问题类似？之前我在1.10.0也遇到过。 http://apache-flink.147419.n8.nabble.com/rocksdb-Could-not-restore-keyed-state-backend-for-KeyedProcessOperator-td2232.html#a2239 解决方式： 1. 使用hdfs作为状态后端不会报错 2. 升级至1.10.1使用rocksdb也不会出现该问题 在 2020-07-14 14:41:53，"Peihui He" 写道： >Hi Yun， >

flink 1.11 sql类型问题

hi， 我看1.11的java.sql.Timestamp 对应的是Flink的TIMESTAMP(9)，跟之前默认的TIMESTAMP(3)有区别，而且之前1.10的Timestamp(3)是带时区UTC的，现在这个类型不带时区了。想问下这个具体调整应该如何适配？

Re: flink1.11 sql kafka 抽取事件时间

我感觉可以通过计算列的方式来解决呀，你只需要在计算rowtime这个列的时候保证它不是null即可，如果是null，可以设置一个默认值之类的？ 18500348...@163.com <18500348...@163.com> 于2020年7月15日周三 下午3:04写道： > 大家好！ > > 使用flink1.11 sql接入kafka ，format为csv > 从eventTime字段中抽取事件时间 > rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(eventTime / 1000, '-MM-dd > HH:mm:ss')) > event

答复: 回复：答复: flink state

Broadcast state是无法满足你的要求的，估计你只能像我这样把涉及的state数据融入到数据流中，在算子中针对不同的类型数据做区分了，等于人工维持这个broadcast的流的变化。 发件人: Robert.Zhang <173603...@qq.com> 日期: 星期三, 2020年7月15日 15:22 收件人: user-zh , user-zh@flink.apache.org 主题: 回复：答复: flink state 是这样的，问题在于我需要使用keyed state 来修改broadcast state，比如根据keyed state把某些满足条件的key存

springboot 2.3.1 + flink 1.11.0整合后如何从外部传入application.yml配置文件?

Hi, 由于项目需要目前整合了springboot和flink, 但一些项目的相关配置在application.yml，生产环境需要变更其中的一些配置。之前通常的做法的是启动时使用 -Dspring.config.location=xxx 从外部指定需要的配置文件。现在想知道如果使用flink run 启动jar能否支持这种指定方式? 谢谢 -- Sent from: http://apache-flink.147419.n8.nabble.com/

?????? pyflink1.11.0window

     ?? --  -- ??: "user-zh"

Re: pyflink1.11.0window

下面这个例子从kafka读取json格式的数据， 然后做窗口聚合后写入es， 可以参考下代码结构， 修改相应数据字段。 这份代码你本地应该是不能运行的 from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic from pyflink.table import StreamTableEnvironment, DataTypes, EnvironmentSettings from pyflink.table.udf import udf @udf(input_types=[DataTypes

Re: Flink 1.11 submit job timed out

Hi Roc 该现象在1.10.1版本没有，在1.11版本才出现。请问这个该如何查比较合适 | | a511955993 | | 邮箱：a511955...@163.com | 签名由 网易邮箱大师 定制 On 07/15/2020 17:16, Roc Marshal wrote: Hi，SmileSmile. 个人之前有遇到过 类似 的host解析问题，可以从k8s的pod节点网络映射角度排查一下。 希望这对你有帮助。 祝好。 Roc Marshal 在 2020-07-15 17:04:18，"SmileSmile" 写道： > >Hi > >使

Re:Flink 1.11 submit job timed out

Hi，SmileSmile. 个人之前有遇到过 类似 的host解析问题，可以从k8s的pod节点网络映射角度排查一下。 希望这对你有帮助。 祝好。 Roc Marshal 在 2020-07-15 17:04:18，"SmileSmile" 写道： > >Hi > >使用版本Flink 1.11，部署方式 kubernetes session。 TM个数30个，每个TM 4个slot。 job >并行度120.提交作业的时候出现大量的No hostname could be resolved for the IP address，JM time >out，作

Re:【求助】Flink Hadoop依赖问题

你好，Z-Z, 可以尝试在 https://repo1.maven.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/ 下载对应的uber jar包，并就将下载后的jar文件放到flink镜像的 ${FLINK_HOME}/lib 路径下，之后启动编排的容器。 祝好。 Roc Marshal. 在 2020-07-15 10:47:39，"Z-Z" 写道： >我在使用Flink 1.11.0版本中，使用docker-compose搭建，docker-compose文件如下： >version: "

Flink 1.11 submit job timed out

Hi 使用版本Flink 1.11，部署方式 kubernetes session。 TM个数30个，每个TM 4个slot。 job 并行度120.提交作业的时候出现大量的No hostname could be resolved for the IP address，JM time out，作业提交失败。web ui也会卡主无响应。 用wordCount，并行度只有1提交也会刷，no hostname的日志会刷个几条，然后正常提交，如果并行度一上去，就会超时。 部分日志如下： 2020-07-15 16:58:46,460 WARN org.apache.flink.

回复: FlinkSQL 入到 MySQL后汉字乱码

是 MySQL_tableB 所在的 server 端字符设置有问题。 配置中加上下面的配置就好了。 [mysqld] character-set-server=utf8 [client] default-character-set=utf8 [mysql] default-character-set=utf8 wangl...@geekplus.com.cn 发件人： wangl...@geekplus.com.cn 发送时间： 2020-07-15 16:34 收件人： user-zh 主题： FlinkSQL 入到 MySQL后汉字乱码 KafkaTable：

Re: flink 1.11 自定义RichFlatMapFunction中使用JdbcRowDataOutputFormat 写pgsql数据问题，RuntimeContext初始化问题，空指针或RuntimeContext未初始化，哪里用的不对！

确实是这行导致的， 如果都重构了，那应该怎么用较好的？ 我需要知道每一行对应的是insert, update还是delete事件。 或者问题变一下，对于这种api，一般遵守什么规则，flink的版本兼容性会更好？ -- Sent from: http://apache-flink.147419.n8.nabble.com/

?????? pyflink1.11.0window

 Shuiqiang      hi2-from_kafka_to_kafka.py??demodemo?? ??

Re: flink 1.9.2 升级 1.10.0 任务失败不能从checkpoint恢复

Hi Robin 其实你的说法不是很准确，社区是明文保证savepoint的兼容性 [1]，但是并不意味着跨大版本时无法从checkpoint恢复，社区不承诺主要还是维护其太耗费精力，但是实际从代码角度来说，在合理使用state schema evolution [2]的前提下，目前跨版本checkpoint恢复基本都是兼容的. 另外 @Peihui 也请麻烦对你的异常描述清晰一些，我的第一次回复已经推测该异常不是root cause，还请在日志中找一下无法恢复的root cause，如果不知道怎么从日志里面找，可以把相关日志分享出来。 [1] https://ci.apac

FlinkSQL 入到 MySQL后汉字乱码

KafkaTable：kafka 消息 MySQL_tableA： 维表，维表里 value 是汉字 MySQL_tableB： join后的结果表。和 MySQL_tableA 不在同一台服务器上。 我直接在 flink sql client SELECT 是可以正常显示， 但 INSERT INTO MySQL_tableB SELECT 后到 MySQL_tableB 里去查看，汉字就乱码了。 大家有什么建议吗？ 谢谢， 王磊 wangl...@geekplus.com.cn

Re: flink 1.9.2 升级 1.10.0 任务失败不能从checkpoint恢复

据我所知，跨大版本的不能直接从checkoint恢复，只能放弃状态重新跑 Best Robin Zhang From: Peihui He <[hidden email]> Sent: Tuesday, July 14, 2020 10:42 To: [hidden email] <[hidden email]> Subject: flink 1.9.2 升级 1.10.0 任务失败不能从checkpoint恢复 hello， 当升级到1.10.0 时候，程序出错后会尝试从checkpoint恢复，但是总

??????????: flink state

keyed state ??broadcast state??keyed statekeybroadcast statebroadcast state??key nonbroadcast side??broadcast state??read-only, ?? -

flink1.11 sql kafka 抽取事件时间

大家好！ 使用flink1.11 sql接入kafka ，format为csv 从eventTime字段中抽取事件时间 rowtime AS TO_TIMESTAMP(FROM_UNIXTIME(eventTime / 1000, '-MM-dd HH:mm:ss')) eventTime可能存在脏数据（非13位的毫秒时间戳），设置了 'csv.ignore-parse-errors' = 'true', 那么eventTime会被设置为null，此时会报一个异常： Caused by: java.lang.RuntimeException: RowTime field shou