hi everyone，
flink version: 1.12.0
job dag: kafka ->hive
今天碰到一个问题，我在第一次启动作业的时候，通过hivecatalog成功在hive中创建hive
table，并正常入数，正常做checkpoint，但由于kafka含有脏数据，导致作业在重启3次仍无法跳过脏数据后变为Failed状态，于是修改作业kafka配置，开启可跳过解析异常行参数，再通过-s
Flink版本 1.12.0
发件人： amen...@163.com
发送时间： 2021-01-03 16:09
收件人： user-zh
主题： Flink SQL DDL Schema csv嵌套json
hi everyone,
zhangsan|man|28|goodst...@gmail.com|{"math":98, "language":{"english":89,
"french":95}}|china|beijing
这是一条来自kafka消息队列中的数据，当我
hi everyone,
zhangsan|man|28|goodst...@gmail.com|{"math":98, "language":{"english":89,
"french":95}}|china|beijing
这是一条来自kafka消息队列中的数据，当我创建kafka ddl为之定义schema时，报出异常信息：
Exception in thread "main" java.lang.IllegalArgumentException: Only simple
types are supported in the second level nesting of
想请问下，写filesystem的时候依赖checkpoint进行commit，那么做完一次checkpoint的时候可提交的文件数是由并行度parallelism数决定的吗？我发现我的文件提交数都是3个3个的当每次chk结束后。
发件人： amen...@163.com
发送时间： 2020-12-24 18:47
收件人： user-zh
主题： Re: Re: Flink-1.11.1流写filesystem分区提交问题
一语点醒梦中人，谢谢回复@冯嘉伟
因为我是先在sql-client中进行的提交测试，因此忽略了这个问题，谢谢
best,
amenhub
一语点醒梦中人，谢谢回复@冯嘉伟
因为我是先在sql-client中进行的提交测试，因此忽略了这个问题，谢谢
best,
amenhub
发件人： 冯嘉伟
发送时间： 2020-12-24 18:39
收件人： user-zh
主题： Re: Flink-1.11.1流写filesystem分区提交问题
有开启checkpoint吗？
Part files can be in one of three states:
In-progress : The part file that is currently being written to is
完了，现在的问题是发现好像所有的分区都没有提交，一直不提交，这是为什么呢？
发件人： amen...@163.com
发送时间： 2020-12-24 17:04
收件人： user-zh
主题： Flink-1.11.1流写filesystem分区提交问题
hi everyone,
最近在验证需求，kafka数据流写hdfs文件系统，使用官网文档Flink-1.11版本的示例demo成功提交到yarn之后，发现如期生成分区目录及文件，但是分区提交有些疑惑想请教下大家。
问题描述：
在15点37分时，查看hdfs如期生成[/user/flink/order/dt=2020
hi everyone,
最近在验证需求，kafka数据流写hdfs文件系统，使用官网文档Flink-1.11版本的示例demo成功提交到yarn之后，发现如期生成分区目录及文件，但是分区提交有些疑惑想请教下大家。
问题描述：
这个问题应该问yarn吧。。。
发件人： yujianbo
发送时间： 2020-12-21 16:43
收件人： user-zh
主题： Flink on yarn 如何指定固定几台yarn节点当做flink任务的运行节点
各位大佬好：
请问Flink on yarn 如何指定固定几台yarn节点当做flink任务的运行节点？
--
Sent from: http://apache-flink.147419.n8.nabble.com/
今天又博学了，谢谢！
发件人： Leonard Xu
发送时间： 2020-12-21 15:01
收件人： user-zh
主题： Re: Flink-1.12支持kafka join jdbc维表吗
>
> 这么讲的话算是比较清晰了，所以如果想要基于事件时间进行jdbc维表Join，首先需要将jdbc维表的changelog数据接入kafka再进行join，这也是官网给的例子，对吗？
是的
> 你说的这种方式就是好像基于处理时间的join~
是的，基于处理时间的维表join和大家熟知的lookup关联,
感谢@Leonard Xu 的回复，
这么讲的话算是比较清晰了，所以如果想要基于事件时间进行jdbc维表Join，首先需要将jdbc维表的changelog数据接入kafka再进行join，这也是官网给的例子，对吗？
>>> 另外，如果对维表实时性要求不高，也没有延迟join的需求，你也可以用运行时lookup的方式完成维表join（即：JOIN latest_rates
>>> FOR SYSTEM_TIME AS OF o.proctime），这种方式是在每条主流的数据到来时去查询一次外部DB，这种方式支持你参数中写的
>>> 'lookup.cache.max-rows'
hi,
请问kafka join jdbc维表数据而不是join jdbc的changelog，支持吗？
在给jdbc声明主键和watermark后，使用官网的事件时间join语法，任务启动后发现jdbc算子一次性读取了所有jdbc维表数据，状态变成了finished，这样的话按理来说不管维表数据怎么变kafka都join不到维表数据了呀？
CREATE TABLE orders (
order_id STRING,
currency STRING,
amount INT,
order_time TIMESTAMP(3),
WATERMARK
hi everyone,
使用./bin/flink stop -yid application_xxx_xxx
xx命令停止yarn任务时，可否将kerberos认证信息带过去？
例如说，-Dsecurity.kerberos.login.keytab 及
-Dsecurity.kerberos.login.principal这样（这二者试过不行 ）
btw, 代码形式的stop和命令行的stop命令使用效率是一致的吗？thanks.
best,
amenhub
hi,
可以在调用executeSql()或execute()（多SQL）的时候使用ugi认证进行任务提交，以及在run命令中动态传入你所说的两个参数，以-yD的形式。
best,
amenhub
发件人： zhengmao776
发送时间： 2020-12-03 17:16
收件人： user-zh
主题： flink使用多个keytab
你好，我在使用flink
execution.attached的值是true，execution.target值是yarn-per-job
best,
amenhub
发件人： amen...@163.com
发送时间： 2020-11-13 11:30
收件人： user-zh
主题： Re: Re: Flink与Yarn的状态一致性问题
hi
1.确定提交的是Per-job模式，提交命令是./bin/flink run -m yarn-cluster xxx，并且从Flink web ui中的Job
Manager -> Configuration观察到execution.target值为yarn-
execution.attached的值是true，execution.target值是yarn-per-job
best,
amenhub
发件人： amen...@163.com
发送时间： 2020-11-13 11:30
收件人： user-zh
主题： Re: Re: Flink与Yarn的状态一致性问题
hi
1.确定提交的是Per-job模式，提交命令是./bin/flink run -m yarn-cluster xxx，并且从Flink web ui中的Job
Manager -> Configuration观察到execution.target值为yarn-
hi
1.确定提交的是Per-job模式，提交命令是./bin/flink run -m yarn-cluster xxx，并且从Flink web ui中的Job
Manager -> Configuration观察到execution.target值为yarn-per-job
2.整体任务状态为Failed，但是TM挂了，JM没有挂（没有挂的原因猜测是因为Yarn application还在Running的原因吧？所以还能从Job
Manager -> logs查看失败日志内容）
best,
amenhub
发件人： JasonLee
发送时间： 2020-11-13
hi
1.确定提交的是Per-job模式，提交命令是./bin/flink run -m yarn-cluster xxx，并且从Flink web ui中的Job
Manager -> Configuration观察到execution.target值为yarn-per-job
2.整体任务状态为Failed，但是TM挂了，JM没有挂（没有挂的原因猜测是因为Yarn application还在Running的原因吧？所以还能从Job
Manager -> logs查看失败日志内容）
best,
amenhub
发件人： JasonLee
发送时间： 2020-11-13
>>>当然，在 FLINK 察觉自己 FAILED 到上报给 YARN 是有一定的时延的，也有可能因为网络等问题上报失败。
按照这个说法，应当是偶发性行为，然而我一直等待Flink上报，大概几个小时过去了Yarn状态仍然处于Running..
>>>你这个是短暂的不一致时间窗口，还是说 FLINK 集群已经退了，YARN 的状态还没有变化呢？
这个话没有看懂，我的提交方式是./bin/flink run -m yarn-cluster xxx，Flink版本是1.11.1
昨天在社区邮件里发现了Flink-1.10以前可以通过-d参数解决Per-job模式下Flink web
hi,
我现在的版本是flink-1.11.1没有加-d参数，也遇见了同样的问题，不知道是什么情况呢？
best,
amenhub
发件人： Yang Wang
发送时间： 2020-08-05 10:28
收件人： user-zh
主题： Re: flink1.9.1任务已经fail掉了，但在yarn上这个application还是在running
你的Flink任务应该是用attach的方式起的，也就是没有加-d，这种情况在1.10之前起的任务本质上是一个session，
hi everyone,
最近在使用Flink-1.11.1 On Yarn Per
Job模式提交简单的kafka->mysql任务时，发现当有脏数据或mysql主键限制等等原因导致Flink任务状态置为Failed时，Yarn
application仍处于运行状态
疑问是Flink任务处于Failed或Finished时，不会反馈自身状态给Yarn吗？期待大佬解惑，谢谢
best,
amenhub
hi everyone,
Flink SQL有没有上一个SQL的输出是下一个SQL的输入的业务场景思路？
比如说KafkaSource -> SQL_1 -> SQL_2 -> MysqlSink，一整个链起来，作为一个任务提交运行~
best,
amenhub
hi everyone,
最近在使用Flink-1.11.1 On Yarn Per
Job模式提交简单的kafka->mysql任务时，发现当有脏数据或mysql主键限制等等原因导致Flink任务状态置为Failed时，Yarn
application仍处于运行状态
疑问是Flink任务处于Failed或Finished时，不会反馈自身状态给Yarn吗？期待大佬解惑，谢谢
best,
amenhub
hi everyone,
想请问下社区及各位大神们，通过Flink Table
API连接带有kerberos认证的kafka时，怎么做的kafka集群中topic和group权限认证的？
best,
amenhub
hi everyone,
最近使用flink-1.11.1在通过per-job方式提交任务到yarn队列的时候，碰到了kerberos权限认证问题。
具体描述：在客户端通过Kerberos权限认证指定用户，提交flink任务到yarn队列，正常提交，但是当任务被yarn分配到指定节点进行执行时，根据报错信息来看，是因为需要操作hdfs(创建检查点目录和保存点目录，因为我使用FileSystem
StateBackend)而并没有获得操作hdfs的权限，被kerberos常规的拦截了。
是的，正如@chenxuying 和@zhisheng 所说，
我这边采用的方案是通过pipeline.classpaths参数将需要的udf jar添加到类路径中，但是当task被分配到tm去执行时仍需要找到所需udf
jar才行，所以在1.11版本中我采用-yt参数将/plugins插件目录上传至hdfs，即可解决这个问题~
best,
amenhub
发件人： zhisheng
发送时间： 2020-10-22 23:28
收件人： user-zh
主题： Re: flink1.11加载外部jar包进行UDF注册
hi
flink 1.11 如果是要管理 udf
还真是不支持，多谢解惑~
发件人： Peidian Li
发送时间： 2020-10-22 19:13
收件人： user-zh
主题： Re: Flink-1.11.1 Rest API使用
Yarn 的proxy server不支持POST请求，这是前几天yarn同事给我截的图：
我们改了下proxy server的逻辑来支持POST请求就可以了
一个stop with savepoint例子：
hi everyone,
如[1]所示的Rest API该怎么进行参数设置并在postman中测试呢？疑惑的地方在于其中的Query
Parameter参数例如mode该怎么传给URL？如果有例子说明，感激不尽
[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/monitoring/rest_api.html#jobs-jobid-1
best,
amenhub
是的，同款TEMPORARY FUNCTION错误，但是使用SYSTEMTEMPORARY就没有问题，不知是否是flink的bug
best,
amenhub
发件人： 史 正超
发送时间： 2020-10-16 15:26
收件人： user-zh@flink.apache.org
主题： 回复: 回复：回复: flink 自定义udf注册后不能使用
你这样创建试一下，或者换个名字试试
CREATE TEMPORARY SYSTEM FUNCTION imei_encrypt AS
'com.intsig.flink.udf.IMEIEncrypt' LANGUAGE
追加问题，在使用线程上下文类加载器的时候，数据会重复发送三条，这是因为添加pipeline.classpaths的缘故吗？
那这种设置env的方式有可能还会造成其他什么问题？
best,
amenhub
发件人： amen...@163.com
发送时间： 2020-10-15 19:22
收件人： user-zh
主题： Re: Re: flink1.11加载外部jar包进行UDF注册
非常感谢您的回复！
对于我来说，这是一个非常好的办法，但是我有一个疑问：类加载器只能使用系统类加载器进行加载吗？
因为我在尝试使用系统类加载器的时候，本身作业包开放给外部UDF
jar包实现的
非常感谢您的回复！
对于我来说，这是一个非常好的办法，但是我有一个疑问：类加载器只能使用系统类加载器进行加载吗？
因为我在尝试使用系统类加载器的时候，本身作业包开放给外部UDF
jar包实现的接口会报ClassNotFound异常，而将类加载器指向主类（这种方式的话这里应该是使用默认的线程上下文加载器），则可避免这个问题。
期待您的回复，谢谢~
best,
amenhub
发件人： cxydeve...@163.com
发送时间： 2020-10-15 17:46
收件人： user-zh
主题： Re: flink1.11加载外部jar包进行UDF注册
hi, everyone
近期有做一个关于从外部路径加载UDF的开发，但报了如下异常：(截取主要的异常信息)
org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot load user
class: com.xxx.xxx.udf.Uppercase
ClassLoader info: URL ClassLoader:
file:
如果是mysql5.x以上的版本，url中autoReconnect参数会无效吧，
可以尝试下修改配置文件wait_timeout/interactive_out参数
best,
amenhub
发件人： 酷酷的浑蛋
发送时间： 2020-08-31 20:48
收件人： user-zh@flink.apache.org
主题： 回复： flink1.11连接mysql问题
下面是我连接mysql的配置，用的flink-1.11.1，还是报那个错误
CREATE TABLE xx(
`xx` varchar,
`xx` varchar
) WITH (
hi Jian Wang,
根据我的理解，在flink
lib目录下导入官方的flink-sql-connector-hive-2.2.0_2.11-1.11.1.jar是指hive[2.0.0-2.2.0]版本都可以使用此依赖。
关于你的问题我曾经遇到过，hive也是2.1.1，我的demo参考[1]可以运行成功，而不需要额外导入flink-sql-connector-hive-2.2.0_2.11-1.11.1.jar，
只需要把[1]中的依赖改成provided并把其jar包导入flink/lib即可。
希望能帮到你，
[1]
hi, everyone
当我把jar包都上传至hdfs时，使用如下命令进行application mode提交，
./bin/flink run-application -t yarn-application
-Dyarn.provided.lib.dirs="hdfs:///user/flink/lib" -c
com.yui.flink.demo.Kafka2Mysql hdfs:///user/flink/app_jars/kafka2mysql.jar
报异常如下：
/deployment/hadoop.html#providing-hadoop-classes
On Mon, Aug 24, 2020 at 9:05 PM amen...@163.com wrote:
>
> 补充一下，当我移除hive-exec等程序中的hadoop依赖时，任务依旧异常，所以也许是我哪个地方没有到位，觉得依赖冲突是因为在测试hive集成之前，我提交过到yarn执行并无异常，所以排查思路来到了hive这里，
> 现在看来，可能是另外某个原因导致的，贴一点点异常栈如下：
&
)
at
org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:424)
... 24 more
best,
amenhub
发件人： amen...@163.com
发送时间： 2020-08-24 20:40
收件人： user-zh
主题： hive-exec依赖导致hadoop冲突问题
hi, everyone
组件版本：flink-1.11.1，hive-2.1.1
问题描述：
使用Table
API调用executeSql()方法编写kafka2mysql实时程序demo，在未导入hive
hi, everyone
组件版本：flink-1.11.1，hive-2.1.1
问题描述：
使用Table
API调用executeSql()方法编写kafka2mysql实时程序demo，在未导入hive-exec依赖时，打包提交到yarn集群，正常运行；
当测试HiveCatalog及读写Hive Table时，Standalone Cluster运行无异常，在flink端正常读写hive
table（不会发生hadoop依赖冲突）；
多谢！已关注~
Best
amen...@163.com
发件人： Leonard Xu
发送时间： 2020-07-24 16:20
收件人： user-zh
主题： Re: flink 1.11 cdc相关问题
Hi amenhub
针对这个问题，我建了个issue来跟踪这个问题[1],
另外你可以在你的PG 里面把表的IDENTITY设置为FULL，这样 debezium 同步的UPDATE数据就会有完整的信息，
DB命令是：ALTER TABLE yourTable REPLICA IDENTITY FULL, 可以参考debezium官网文档[2]
Best
感谢二位大佬@Leonard, @Jark的解答！
amen...@163.com
发件人： Jark Wu
发送时间： 2020-07-22 23:56
收件人： user-zh
主题： Re: flink 1.11 cdc相关问题
Hi,
这是个已知问题，目前 debezium 同步不同数据库并没有保证一模一样地数据格式，比如同步 PG 的UPDATE消息时候，before 和
after 字段就不是全的。
这个问题会在后面地版本中解决。
Best,
Jark
On Wed, 22 Jul 2020 at 21:07, Leonard Xu wrote
tors.StreamSource.run(StreamSource.java:63)
~[flink-dist_2.11-1.11.0.jar:1.11.0]
at
org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201)
~[flink-dist_2.11-1.11.0.jar:1.11.0]
Caused by: java.lang.NullPointerException
at
org.apache.flink.formats.json.debezium.DebeziumJsonDeserializationSchema.deserialize(DebeziumJsonDeserializationSchema.java:120)
~[flink-json-1.11.0.jar:1.11.0]
... 7 more
2020-07-22 17:22:34,415 INFO org.apache.flink.runtime.taskmanager.Task [] -
Freeing task resources for Source: TableSourceScan(table=[[default_catalog,
default_database, pgsql_person_cdc]], fields=[id, name, age, sex, phone]) ->
Sink: Sink(table=[default_catalog.default_database.sql_out], fields=[id, name,
age, sex, phone]) (1/1) (b553cb66df6e47a27e7dae8466b684ab).
2020-07-22 17:22:34,418 INFO org.apache.flink.runtime.taskexecutor.TaskExecutor
[] - Un-registering task and sending final execution state FAILED to JobManager
for task Source: TableSourceScan(table=[[default_catalog, default_database,
pgsql_person_cdc]], fields=[id, name, age, sex, phone]) -> Sink:
Sink(table=[default_catalog.default_database.sql_out], fields=[id, name, age,
sex, phone]) (1/1) b553cb66df6e47a27e7dae8466b684ab.
2020-07-22 17:22:34,461 INFO
org.apache.flink.runtime.taskexecutor.slot.TaskSlotTableImpl [] - Free slot
TaskSlot(index:0, state:ACTIVE, resource profile:
ResourceProfile{cpuCores=1., taskHeapMemory=384.000mb
(402653174 bytes), taskOffHeapMemory=0 bytes, managedMemory=512.000mb
(536870920 bytes), networkMemory=128.000mb (134217730 bytes)}, allocationId:
495bb5a0cd877808674b29890b6b8bc0, jobId: 3feda3a191fcb8e0da891b9fda1ee532).
2020-07-22 17:22:34,462 INFO
org.apache.flink.runtime.taskexecutor.DefaultJobLeaderService [] - Remove job
3feda3a191fcb8e0da891b9fda1ee532 from job leader monitoring.
2020-07-22 17:22:34,462 INFO org.apache.flink.runtime.taskexecutor.TaskExecutor
[] - Close JobManager connection for job 3feda3a191fcb8e0da891b9fda1ee532.
分割线==
best!
amen...@163.com
hi, everyone
环境信息：flink-1.11.0, blink-planner, 本地ide开发测试（IDEA）
问题描述：使用executeSql()方法执行DDL语句，控制台打印如下异常信息。（flink-connector-kafka_2.11依赖已添加）
我不确定是否还有某个必要的依赖没有添加，还是有其他的地方没有考虑完整，请大佬赐教。
你好，
社区已经发现过一些与 metaspace
泄漏相关的问题，其中部分已经修复，还有一些则与第三方依赖有关。由于你的图片没有显示出来，我不确定你遇到的是否是相同的问题。
此外，在即将发布的 1.11 中，Flink 针对这一问题进行了优化，作业会采用单独的 ClassLoader 运行，以避免作业结束之后
metaspace 还有泄漏的问题。1.11.0 已经进入发布前的测试阶段，RC1 已经发布，欢迎试用。
Thank you~
Xintong Song
On Fri, Jun 5, 2020 at 9:29 AM 胡泽康 wrote:
> flink版本：1.10
hi，了解到使用stop进行任务停止并触发savepoint，会在停止之前生成max_watermark，并注册event-time计时器，我想请问使用yarn
kill方式直接停止任务，会属于cancel还是stop亦或是其他？
amen...@163.com
From: Congxian Qiu
Date: 2020-01-10 17:16
To: user-zh
Subject: Re: flink savepoint checkpoint
Hi
从 Flink 的角度看，Checkpoint 用户 Job 运行过程中发生 failover 进行恢复，savepoint
43 matches
Mail list logo
