date:20210830

Flink on yarn的日志监控和checkpoint的监控生产是如何处理的？

2021-08-30 文章 guanyq

flink on yarn 在集群中启动很多的task，生产应用中是如何监控task的日志，和checkpoint的呢？


求大佬指导。

Re: flink run -d -m yarn-cluster 提交任务到yarn集群不成功

2021-08-30 文章 Yang Wang

export HADOOP_CLASSPATH=`hadoop classpath`

如上方式应该是没有问题的，你确认下这些目录下面的jar包是存在的，尤其是/Users//local/hadoop/hadoop-3.2.2/share/hadoop/yarn/

Best,
Yang

龙逸尘  于2021年8月31日周二 上午11:02写道：

> Hi Wayne，
>
> 可以尝试下指定 HADOOP_CONF_DIR
> export HADOOP_CONF_DIR=/opt/flink/hadoop-conf/
>
> Wayne <1...@163.com> 于2021年8月28日周六 下午8:37写道：
>
> > 我的提交命令
> >
> >
> > ./bin/flink run -d -m yarn-cluster
> >
> >
> > 报错如下
> >  The program finished with the following exception:
> >
> >
> > java.lang.IllegalStateException: No Executor found. Please make sure to
> > export the HADOOP_CLASSPATH environment variable or have hadoop in your
> > classpath. For more information refer to the "Deployment" section of the
> > official Apache Flink documentation.
> > at
> >
> org.apache.flink.yarn.cli.FallbackYarnSessionCli.isActive(FallbackYarnSessionCli.java:41)
> > at
> >
> org.apache.flink.client.cli.CliFrontend.validateAndGetActiveCommandLine(CliFrontend.java:1236)
> > at
> > org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:234)
> > at
> >
> org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
> > at
> >
> org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
> > at
> >
> org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
> > at
> > org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
> >
> >
> > 运行命令 hadoop classpath
> > @192 flink-1.12.2 % hadoop classpath
> >
> >
> /Users//local/hadoop/hadoop-3.2.2/etc/hadoop:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/common/lib/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/common/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/hdfs:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/hdfs/lib/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/hdfs/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/mapreduce/lib/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/mapreduce/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/yarn:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/yarn/lib/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/yarn/*:/Users//local/hadoop/hadoop-3.2.2
> > 反复配置HADOOP_CLASSPATH 无法生效 官网给出的
> > export HADOOP_CLASSPATH=`hadoop classpath`
> > 这个 hadoop classpath 具体配置到哪一级
> >
> >
> >
> >
> >
> >
> >
> >
>

Re: flink run -d -m yarn-cluster 提交任务到yarn集群不成功

2021-08-30 文章龙逸尘

Hi Wayne，

可以尝试下指定 HADOOP_CONF_DIR
export HADOOP_CONF_DIR=/opt/flink/hadoop-conf/

Wayne <1...@163.com> 于2021年8月28日周六 下午8:37写道：

> 我的提交命令
>
>
> ./bin/flink run -d -m yarn-cluster
>
>
> 报错如下
>  The program finished with the following exception:
>
>
> java.lang.IllegalStateException: No Executor found. Please make sure to
> export the HADOOP_CLASSPATH environment variable or have hadoop in your
> classpath. For more information refer to the "Deployment" section of the
> official Apache Flink documentation.
> at
> org.apache.flink.yarn.cli.FallbackYarnSessionCli.isActive(FallbackYarnSessionCli.java:41)
> at
> org.apache.flink.client.cli.CliFrontend.validateAndGetActiveCommandLine(CliFrontend.java:1236)
> at
> org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:234)
> at
> org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
> at
> org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
> at
> org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
> at
> org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
>
>
> 运行命令 hadoop classpath
> @192 flink-1.12.2 % hadoop classpath
>
> /Users//local/hadoop/hadoop-3.2.2/etc/hadoop:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/common/lib/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/common/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/hdfs:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/hdfs/lib/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/hdfs/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/mapreduce/lib/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/mapreduce/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/yarn:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/yarn/lib/*:/Users//local/hadoop/hadoop-3.2.2/share/hadoop/yarn/*:/Users//local/hadoop/hadoop-3.2.2
> 反复配置HADOOP_CLASSPATH 无法生效 官网给出的
> export HADOOP_CLASSPATH=`hadoop classpath`
> 这个 hadoop classpath 具体配置到哪一级
>
>
>
>
>
>
>
>

Re: flink oss ha

2021-08-30 文章 Yun Tang

Hi

这个看上去更像是oss配置的问题，你能使用目前配置的 oss.endpoint，accessKeyId以及accessKeySecret 
结合ossutil等工具访问对应的 oss://bucket-logcenter/flink-state/flink-session-recovery 么？

祝好
唐云

From: dker eandei 
Sent: Monday, August 30, 2021 12:36
To: user-zh@flink.apache.org 
Subject: 回复: flink oss ha

您好:
 附件是使用oss作高可用时的报错，以下是启动flink时的脚本:

../bin/kubernetes-session.sh \
-Dkubernetes.cluster-id=flink-session-1 \
-Dkubernetes.container.image=test/flink:1.13.2-scala_2.12-oss \
-Dkubernetes.container.image.pull-policy=Always \
-Dkubernetes.namespace=flink-session \
-Dkubernetes.service-account=flink-session-sa \
-Dkubernetes.rest-service.exposed.type=ClusterIP \
-Dtaskmanager.numberOfTaskSlots=6 \
-Djobmanager.memory.process.size=1024m \
-Dtaskmanager.memory.process.size=2048m \
-Dkubernetes.jobmanager.cpu=1 \
-Dkubernetes.taskmanager.cpu=2 \
-Dfs.oss.endpoint="http://oss-.local; \
-Dfs.oss.accessKeyId="j0BAJ" \
-Dfs.oss.accessKeySecret="7mzTPiC4w" \

-Dhigh-availability=org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
 \

-Dhigh-availability.storageDir=oss://bucket-logcenter/flink-state/flink-session-recovery
 \

-Dcontainerized.master.env.ENABLE_BUILT_IN_PLUGINS=flink-oss-fs-hadoop-1.13.2.jar
 \

-Dcontainerized.taskmanager.env.ENABLE_BUILT_IN_PLUGINS=flink-oss-fs-hadoop-1.13.2.jar

-邮件原件-
发件人: Yun Tang 
发送时间: 2021年8月30日 11:36
收件人: user-zh@flink.apache.org
主题: Re: flink oss ha

Hi,
你好，图片无法加载，可以直接粘贴文字出来

祝好
唐云

From: dker eandei 
Sent: Friday, August 27, 2021 14:58
To: user-zh@flink.apache.org 
Subject: flink oss ha


您好:

看文档OSS可以用作 FsStatebackend，那么Flink on k8s 
做高可用时，high-availability.storageDir可以配置成oss吗，我试了下，报以下错误:

[cid:image002.png@01D79B53.F4C71E80]



从 Windows 
版邮件发送

Re: flink run 提交任务到yarn 报Cannot load user class: org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

2021-08-30 文章 Caizhi Weng

Hi！

-C 要求提供的路径必须在集群的所有节点都能访问。如果提供的是 file://，那么集群的所有节点的这个路径下都要有对应文件。可以把文件放在 hdfs
上然后 -C 指定 hdfs:// 试试。

Wayne <1...@163.com> 于2021年8月30日周一 下午3:13写道：

> 我的执行命令是
>  flink run -m yarn-cluster -yd -yjm 1024m -ytm 1024m -ys 1 -ynm xxx -C
> file:///xxx/flink-connector-kafka_2.12-1.12.2.jar -C
> file:///xxx/flink-sql-avro-1.12.2.jar ...
> 提交到生产集群上提示
> org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot load
> user class: org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
> ClassLoader info: URL ClassLoader:
> Class not resolvable through given classloader.
> at
> org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:331)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain.(OperatorChain.java:145)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:517)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_292]
> Caused by: java.lang.ClassNotFoundException:
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
> at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
> ~[?:1.8.0_292]
> at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
> ~[?:1.8.0_292]
> at
> org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:64)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
> ~[?:1.8.0_292]
> at
> org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:172)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at java.lang.Class.forName0(Native Method) ~[?:1.8.0_292]
> at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_292]
> at
> org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:76)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1986)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1850)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2160)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readObject(ObjectInputStream.java:503)
> ~[?:1.8.0_292]
> at
> java.io.ObjectInputStream.readObject(ObjectInputStream.java:461)
> ~[?:1.8.0_292]
> at
> org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:600)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:587)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:541)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> at
> org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:317)
> ~[flink-dist_2.12-1.12.2.jar:1.12.2]
> ... 6 more
> 2
>
>
> 任务可以提交到yarn上，说明 -C

退订

2021-08-30 文章 xiaobo77

退订

??????flink keyby????

2021-08-30 文章 JasonLee

hi


 KeyGroupStreamPartitioner#selectChannel .


Best
JasonLee
??2021??8??30?? 22:34??cs<58683...@qq.com.INVALID> ??
flink??keybykey??tasktask??id

flink keyby????

2021-08-30 文章 cs

flink??keybykey??tasktask??id

Re: Flink 从checkpoint恢复时，部分状态没有正确恢复

2021-08-30 文章 Benchao Li

这个问题已经在1.12中修复了，参考:
https://issues.apache.org/jira/browse/FLINK-18688

Benchao Li  于2021年8月30日周一 下午7:49写道：

> Hi xingxing，
>
> 看起来你可能也遇到了这个bug了。
> 我们遇到过一个bug是这样的，在group by的多个字段里面，如果有2个及以上变长字段的话，会导致底层的BinaryRow序列化
> 的结果不稳定，进而导致状态恢复会错误。
> 先解释下变长字段，这个指的是4byte存不下的数据类型，比如典型的varchar、list、map、row等等；
> 然后再解释下这个结果不稳定的原因，这个是因为在底层的代码生成里面有一个优化，会按照类型进行分组，然后进行优化，
> 但是这个分组的过程用的是一个HashMap[1]，会导致字段顺序不是确定性的，有时候是这个顺序，有时候又是另外一个顺序，
> 导致最终的BinaryRow的序列化结果是不稳定的，进而导致无法从checkpoint恢复。
>
> 然后典型的多种变长类型，其实是varchar nullable 和 varchar not null 以及
> char(n)，尤其是你这种用了很多常量字符串的场景，
> 容易产生后两种类型，在加上普通字段以及函数都会产生的第一种类型，就会触发这个bug了。
>
> [1]
> https://github.com/apache/flink/blob/release-1.9/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala#L92
>
> dixingxing  于2021年8月25日周三 下午8:57写道：
>
>> Hi Flink 社区：
>> 我们的Flink版本是1.9.2，用的是blink planer，我们今天遇到一个问题，目前没有定位到原因，现象如下：
>>
>> 手动重启任务时，指定了从一个固定的checkpoint恢复时，有一定的概率，一部分状态数据无法正常恢复，启动后Flink任务本身可以正常运行，且日志中没有明显的报错信息。
>> 具体现象是：type=realshow的数据没有从状态恢复，也就是从0开始累加，而type=show和type=click的数据是正常从状态恢复的。
>>
>>
>> SQL大致如下：
>> createview view1 as
>> select event_id, act_time, device_id
>> from table1
>> where `getStringFromJson`(`act_argv`, 'ispin', '') <>'1'
>> and event_id in
>> ('article_newest_list_show','article_newest_list_sight_show',
>> 'article_list_item_click', 'article_auto_video_play_click');
>>
>>
>> --天的数据
>> insertinto table2
>> select platform, type, `time`, count(1) as pv, hll_uv(device_id) as uv
>> from
>> (select'03'as platform, trim(casewhen event_id
>> ='article_newest_list_show'then'show'
>> when event_id ='article_newest_list_sight_show'then'realshow'
>> when event_id ='article_list_item_click'then'click'else''end) astype,
>> `date_parse`(`act_time`, '-MM-dd HH:mm:ss', 'MMdd') as `time`,
>> device_id
>> from view1
>> where event_id in
>> ('article_newest_list_show','article_newest_list_sight_show',
>> 'article_list_item_click')
>> unionall
>> select'03'as platform, 'click_total'astype,
>> `date_parse`(`act_time`, '-MM-dd HH:mm:ss', 'MMdd') as `time`,
>> device_id
>> from view1
>> where event_id in ('article_list_item_click',
>> 'article_auto_video_play_click'))a
>> groupby platform, type, `time`;
>>
>>
>> 期待大家的帮助与回复，希望能给些问题排查的思路！
>>
>>
>>
>>
>
> --
>
> Best,
> Benchao Li
>


-- 

Best,
Benchao Li

Re: Flink 从checkpoint恢复时，部分状态没有正确恢复

2021-08-30 文章 Benchao Li

Hi xingxing，

看起来你可能也遇到了这个bug了。
我们遇到过一个bug是这样的，在group by的多个字段里面，如果有2个及以上变长字段的话，会导致底层的BinaryRow序列化
的结果不稳定，进而导致状态恢复会错误。
先解释下变长字段，这个指的是4byte存不下的数据类型，比如典型的varchar、list、map、row等等；
然后再解释下这个结果不稳定的原因，这个是因为在底层的代码生成里面有一个优化，会按照类型进行分组，然后进行优化，
但是这个分组的过程用的是一个HashMap[1]，会导致字段顺序不是确定性的，有时候是这个顺序，有时候又是另外一个顺序，
导致最终的BinaryRow的序列化结果是不稳定的，进而导致无法从checkpoint恢复。

然后典型的多种变长类型，其实是varchar nullable 和 varchar not null 以及
char(n)，尤其是你这种用了很多常量字符串的场景，
容易产生后两种类型，在加上普通字段以及函数都会产生的第一种类型，就会触发这个bug了。

[1]
https://github.com/apache/flink/blob/release-1.9/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/codegen/ProjectionCodeGenerator.scala#L92

dixingxing  于2021年8月25日周三 下午8:57写道：

> Hi Flink 社区：
> 我们的Flink版本是1.9.2，用的是blink planer，我们今天遇到一个问题，目前没有定位到原因，现象如下：
>
> 手动重启任务时，指定了从一个固定的checkpoint恢复时，有一定的概率，一部分状态数据无法正常恢复，启动后Flink任务本身可以正常运行，且日志中没有明显的报错信息。
> 具体现象是：type=realshow的数据没有从状态恢复，也就是从0开始累加，而type=show和type=click的数据是正常从状态恢复的。
>
>
> SQL大致如下：
> createview view1 as
> select event_id, act_time, device_id
> from table1
> where `getStringFromJson`(`act_argv`, 'ispin', '') <>'1'
> and event_id in
> ('article_newest_list_show','article_newest_list_sight_show',
> 'article_list_item_click', 'article_auto_video_play_click');
>
>
> --天的数据
> insertinto table2
> select platform, type, `time`, count(1) as pv, hll_uv(device_id) as uv
> from
> (select'03'as platform, trim(casewhen event_id
> ='article_newest_list_show'then'show'
> when event_id ='article_newest_list_sight_show'then'realshow'
> when event_id ='article_list_item_click'then'click'else''end) astype,
> `date_parse`(`act_time`, '-MM-dd HH:mm:ss', 'MMdd') as `time`,
> device_id
> from view1
> where event_id in
> ('article_newest_list_show','article_newest_list_sight_show',
> 'article_list_item_click')
> unionall
> select'03'as platform, 'click_total'astype,
> `date_parse`(`act_time`, '-MM-dd HH:mm:ss', 'MMdd') as `time`,
> device_id
> from view1
> where event_id in ('article_list_item_click',
> 'article_auto_video_play_click'))a
> groupby platform, type, `time`;
>
>
> 期待大家的帮助与回复，希望能给些问题排查的思路！
>
>
>
>

-- 

Best,
Benchao Li

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

2021-08-30 文章 wukon...@foxmail.com

我目前大概会采用 Shuo Cheng 提到的， 使用先sink 到Mysql， 再启一个任务 cdc mysql 中的表，这样能保证插入成功后的数据。

我目前使用的是flink 1.12 版本 如果是多端sink 比如 sink db 同时sink kafka ，flink 在sink db 
失败，依旧会sink kafka 但是会因为异常，发生tm 重启，会根据自定义重启策略，一直到最后 整个job fail over 掉。



wukon...@foxmail.com
 
发件人： 东东
发送时间： 2021-08-30 16:50
收件人： user-zh
主题： Re:Re: Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
 
 
 
对于Exactly-Once模式下的checkpoint，如果sink端支持两段式事务，应该就可以做到一个sink失败，整个checkpoint失败的。
 
 
不过JDBC sink支持Exactly-Once是1.13.0以后的事情了，建议检查一下你的版本和配置
 
 
 
 
 
 
 
 
 
在 2021-08-30 16:27:24，"wukon...@foxmail.com"  写道：
>Hi: 
> 我理解这种方式， 目前我只是想让先插入数据到Mysql 然后再通过通知到下游，来查询Mysql 进行数据etl 不知道大家如何通过SQL来实现这一逻辑
>
>
>
>wukon...@foxmail.com
> 
>发件人： Shuo Cheng
>发送时间： 2021-08-30 10:19
>收件人： user-zh
>主题： Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
>你好, 你说的这种控制写入的方式在同一个 Flink SQL job 里是无法实现的. 控制数据是否写入某个
>Sink，可以看看是否从逻辑上能在 Sink 前加一个 Filter，从而达到过滤目的；如果 kafka sink 跟 MySQL
>表是一种类似级联的关系, 可以考虑先写入 MySQL, 然后另起一个 Job 用 CDC 方式读 MySQL changelog 再写入
>Kafka sink.
> 
>On 8/26/21, jie han  wrote:
>> HI：
>> 可以尝试下使用flink cdc 的方式写入到第二个kafka里面呀
>>
>> 悟空  于2021年8月26日周四 下午1:54写道：
>>
>>> 我目前用的是flink-connector-kafka_2.11和flink-connector-jdbc_2.11，
>>> 测试时，我把任务启动好之后，把mysql 中的目标表删除 或 删除必要字段，
>>> 之后发送一条kafka数据，会报java.sql.BatchUpdateException 异常，然后重试3次。
>>> 但是接着sink Kafka 是成功的，Kafka端 我开启了'sink.semantic' = 'exactly-once',
>>> 同时下游consumer 使用--isolation-level read_committed
>>> 读取，依旧能成功读取到数据，说明sink
>>> db 失败，但是sink kafka成功，同时flink 本身任务不会挂掉。
>>>
>>>
>>>
>>>
>>> --原始邮件--
>>> 发件人:
>>>   "user-zh"
>>> <
>>> tsreape...@gmail.com;
>>> 发送时间:2021年8月26日(星期四) 中午1:25
>>> 收件人:"user-zh">>
>>> 主题:Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
>>>
>>>
>>>
>>> Hi！
>>>
>>> 如果 statement set 已经包含了多个 insert 语句，那么写入 kafka 表和写入 db 应该是在同一个作业里进行的。如果写入
>>> db
>>> 失败，那么产生的 exception 应该会让作业失败才对。这里 db 写入失败但 kafka 依旧写入是什么样的现象？
>>>
>>> 另外其实可以考虑分成两个作业，第一个作业将数据写入 db，第二个作业从 db 读出数据写入 kafka。关于捕获 db 数据的变化，可以看一下
>>> Flink CDC connector[1]
>>>
>>> [1] https://github.com/ververica/flink-cdc-connectors
>>>
>>> 悟空 >>
>>>  能否详细说下呢，statement set[1] 是什么意思， 我是通过StatementSet.addInsertSql 将多个sql
>>>  加入的，然后执行execute()方法
>>> 
>>> 
>>> 
>>> 
>>>  --nbsp;原始邮件nbsp;--
>>>  发件人:
>>> 
>>> "user-zh"
>>> 
>>> <
>>>  fskm...@gmail.comgt;;
>>>  发送时间:nbsp;2021年8月26日(星期四) 中午12:36
>>>  收件人:nbsp;"user-zh">> 
>>>  主题:nbsp;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
>>> 
>>> 
>>> ;
>>>  悟空 >> 
>>>  gt; hi all:amp;nbsp;
>>>  gt; amp;nbsp; amp;nbsp; 我目前基于flink 1.12 sql 来开发功能，
>>> 目前遇到一个问题， 我现在想实现
>>>  在一个事务里 先将kafka
>>>  gt; 源的数据写入到一张msyql 表中， 写入成功后再写入一张kafka的表里。如果写入db失败，则不写入kafka 。
>>>  gt; amp;nbsp; amp;nbsp;语句类似这种：
>>>  gt; amp;nbsp; amp;nbsp;insert into
>>> db_table_sinkamp;nbsp;select *
>>>  fromamp;nbsp;
>>>  gt; kafka_source_table;
>>>  gt; amp;nbsp; amp;nbsp;insert into kafka_table_sink
>>> select * from
>>>  kafka_source_table;
>>>  gt;
>>>  gt;
>>>  gt; amp;nbsp; 请问flink SQL 有实现方式吗？
>>> 目前经过测试，发现如果db写入失败，但是kafka依旧写入，同时flink
>>>  程序没有挂掉。
>>

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

2021-08-30 文章东东




对于Exactly-Once模式下的checkpoint，如果sink端支持两段式事务，应该就可以做到一个sink失败，整个checkpoint失败的。


不过JDBC sink支持Exactly-Once是1.13.0以后的事情了，建议检查一下你的版本和配置









在 2021-08-30 16:27:24，"wukon...@foxmail.com"  写道：
>Hi: 
> 我理解这种方式， 目前我只是想让先插入数据到Mysql 然后再通过通知到下游，来查询Mysql 进行数据etl 不知道大家如何通过SQL来实现这一逻辑
>
>
>
>wukon...@foxmail.com
> 
>发件人： Shuo Cheng
>发送时间： 2021-08-30 10:19
>收件人： user-zh
>主题： Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
>你好, 你说的这种控制写入的方式在同一个 Flink SQL job 里是无法实现的. 控制数据是否写入某个
>Sink，可以看看是否从逻辑上能在 Sink 前加一个 Filter，从而达到过滤目的；如果 kafka sink 跟 MySQL
>表是一种类似级联的关系, 可以考虑先写入 MySQL, 然后另起一个 Job 用 CDC 方式读 MySQL changelog 再写入
>Kafka sink.
> 
>On 8/26/21, jie han  wrote:
>> HI：
>> 可以尝试下使用flink cdc 的方式写入到第二个kafka里面呀
>>
>> 悟空  于2021年8月26日周四 下午1:54写道：
>>
>>> 我目前用的是flink-connector-kafka_2.11和flink-connector-jdbc_2.11，
>>> 测试时，我把任务启动好之后，把mysql 中的目标表删除 或 删除必要字段，
>>> 之后发送一条kafka数据，会报java.sql.BatchUpdateException 异常，然后重试3次。
>>> 但是接着sink Kafka 是成功的，Kafka端 我开启了'sink.semantic' = 'exactly-once',
>>> 同时下游consumer 使用--isolation-level read_committed
>>> 读取，依旧能成功读取到数据，说明sink
>>> db 失败，但是sink kafka成功，同时flink 本身任务不会挂掉。
>>>
>>>
>>>
>>>
>>> --原始邮件--
>>> 发件人:
>>>   "user-zh"
>>> <
>>> tsreape...@gmail.com;
>>> 发送时间:2021年8月26日(星期四) 中午1:25
>>> 收件人:"user-zh">>
>>> 主题:Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
>>>
>>>
>>>
>>> Hi！
>>>
>>> 如果 statement set 已经包含了多个 insert 语句，那么写入 kafka 表和写入 db 应该是在同一个作业里进行的。如果写入
>>> db
>>> 失败，那么产生的 exception 应该会让作业失败才对。这里 db 写入失败但 kafka 依旧写入是什么样的现象？
>>>
>>> 另外其实可以考虑分成两个作业，第一个作业将数据写入 db，第二个作业从 db 读出数据写入 kafka。关于捕获 db 数据的变化，可以看一下
>>> Flink CDC connector[1]
>>>
>>> [1] https://github.com/ververica/flink-cdc-connectors
>>>
>>> 悟空 >>
>>>  能否详细说下呢，statement set[1] 是什么意思， 我是通过StatementSet.addInsertSql 将多个sql
>>>  加入的，然后执行execute()方法
>>> 
>>> 
>>> 
>>> 
>>>  --nbsp;原始邮件nbsp;--
>>>  发件人:
>>> 
>>> "user-zh"
>>> 
>>> <
>>>  fskm...@gmail.comgt;;
>>>  发送时间:nbsp;2021年8月26日(星期四) 中午12:36
>>>  收件人:nbsp;"user-zh">> 
>>>  主题:nbsp;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
>>> 
>>> 
>>> ;
>>>  悟空 >> 
>>>  gt; hi all:amp;nbsp;
>>>  gt; amp;nbsp; amp;nbsp; 我目前基于flink 1.12 sql 来开发功能，
>>> 目前遇到一个问题， 我现在想实现
>>>  在一个事务里 先将kafka
>>>  gt; 源的数据写入到一张msyql 表中， 写入成功后再写入一张kafka的表里。如果写入db失败，则不写入kafka 。
>>>  gt; amp;nbsp; amp;nbsp;语句类似这种：
>>>  gt; amp;nbsp; amp;nbsp;insert into
>>> db_table_sinkamp;nbsp;select *
>>>  fromamp;nbsp;
>>>  gt; kafka_source_table;
>>>  gt; amp;nbsp; amp;nbsp;insert into kafka_table_sink
>>> select * from
>>>  kafka_source_table;
>>>  gt;
>>>  gt;
>>>  gt; amp;nbsp; 请问flink SQL 有实现方式吗？
>>> 目前经过测试，发现如果db写入失败，但是kafka依旧写入，同时flink
>>>  程序没有挂掉。
>>

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

2021-08-30 文章 wukon...@foxmail.com

Hi: 
 我理解这种方式， 目前我只是想让先插入数据到Mysql 然后再通过通知到下游，来查询Mysql 进行数据etl 不知道大家如何通过SQL来实现这一逻辑



wukon...@foxmail.com
 
发件人： Shuo Cheng
发送时间： 2021-08-30 10:19
收件人： user-zh
主题： Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
你好, 你说的这种控制写入的方式在同一个 Flink SQL job 里是无法实现的. 控制数据是否写入某个
Sink，可以看看是否从逻辑上能在 Sink 前加一个 Filter，从而达到过滤目的；如果 kafka sink 跟 MySQL
表是一种类似级联的关系, 可以考虑先写入 MySQL, 然后另起一个 Job 用 CDC 方式读 MySQL changelog 再写入
Kafka sink.
 
On 8/26/21, jie han  wrote:
> HI：
> 可以尝试下使用flink cdc 的方式写入到第二个kafka里面呀
>
> 悟空  于2021年8月26日周四 下午1:54写道：
>
>> 我目前用的是flink-connector-kafka_2.11和flink-connector-jdbc_2.11，
>> 测试时，我把任务启动好之后，把mysql 中的目标表删除 或 删除必要字段，
>> 之后发送一条kafka数据，会报java.sql.BatchUpdateException 异常，然后重试3次。
>> 但是接着sink Kafka 是成功的，Kafka端 我开启了'sink.semantic' = 'exactly-once',
>> 同时下游consumer 使用--isolation-level read_committed
>> 读取，依旧能成功读取到数据，说明sink
>> db 失败，但是sink kafka成功，同时flink 本身任务不会挂掉。
>>
>>
>>
>>
>> --原始邮件--
>> 发件人:
>>   "user-zh"
>> <
>> tsreape...@gmail.com;
>> 发送时间:2021年8月26日(星期四) 中午1:25
>> 收件人:"user-zh">
>> 主题:Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
>>
>>
>>
>> Hi！
>>
>> 如果 statement set 已经包含了多个 insert 语句，那么写入 kafka 表和写入 db 应该是在同一个作业里进行的。如果写入
>> db
>> 失败，那么产生的 exception 应该会让作业失败才对。这里 db 写入失败但 kafka 依旧写入是什么样的现象？
>>
>> 另外其实可以考虑分成两个作业，第一个作业将数据写入 db，第二个作业从 db 读出数据写入 kafka。关于捕获 db 数据的变化，可以看一下
>> Flink CDC connector[1]
>>
>> [1] https://github.com/ververica/flink-cdc-connectors
>>
>> 悟空 >
>>  能否详细说下呢，statement set[1] 是什么意思， 我是通过StatementSet.addInsertSql 将多个sql
>>  加入的，然后执行execute()方法
>> 
>> 
>> 
>> 
>>  --nbsp;原始邮件nbsp;--
>>  发件人:
>> 
>> "user-zh"
>> 
>> <
>>  fskm...@gmail.comgt;;
>>  发送时间:nbsp;2021年8月26日(星期四) 中午12:36
>>  收件人:nbsp;"user-zh"> 
>>  主题:nbsp;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
>> 
>> 
>> ;
>>  悟空 > 
>>  gt; hi all:amp;nbsp;
>>  gt; amp;nbsp; amp;nbsp; 我目前基于flink 1.12 sql 来开发功能，
>> 目前遇到一个问题， 我现在想实现
>>  在一个事务里 先将kafka
>>  gt; 源的数据写入到一张msyql 表中， 写入成功后再写入一张kafka的表里。如果写入db失败，则不写入kafka 。
>>  gt; amp;nbsp; amp;nbsp;语句类似这种：
>>  gt; amp;nbsp; amp;nbsp;insert into
>> db_table_sinkamp;nbsp;select *
>>  fromamp;nbsp;
>>  gt; kafka_source_table;
>>  gt; amp;nbsp; amp;nbsp;insert into kafka_table_sink
>> select * from
>>  kafka_source_table;
>>  gt;
>>  gt;
>>  gt; amp;nbsp; 请问flink SQL 有实现方式吗？
>> 目前经过测试，发现如果db写入失败，但是kafka依旧写入，同时flink
>>  程序没有挂掉。
>

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

2021-08-30 文章 wukon...@foxmail.com

能具体说下如何实现吗？ 我用cdc 能实现什么，我现在想让两个Insert Sql 保持到一个事务里， 要么全成功，要么全失败，目前查看Flink 文档 
并没有发现相关的解释



wukon...@foxmail.com
 
发件人： jie han
发送时间： 2021-08-26 21:36
收件人： user-zh
主题： Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
HI：
可以尝试下使用flink cdc 的方式写入到第二个kafka里面呀
 
悟空  于2021年8月26日周四 下午1:54写道：
 
> 我目前用的是flink-connector-kafka_2.11和flink-connector-jdbc_2.11，
> 测试时，我把任务启动好之后，把mysql 中的目标表删除 或 删除必要字段，
> 之后发送一条kafka数据，会报java.sql.BatchUpdateException 异常，然后重试3次。
> 但是接着sink Kafka 是成功的，Kafka端 我开启了'sink.semantic' = 'exactly-once',
> 同时下游consumer 使用--isolation-level read_committed 读取，依旧能成功读取到数据，说明sink
> db 失败，但是sink kafka成功，同时flink 本身任务不会挂掉。
>
>
>
>
> --原始邮件--
> 发件人:
>   "user-zh"
> <
> tsreape...@gmail.com;
> 发送时间:2021年8月26日(星期四) 中午1:25
> 收件人:"user-zh"
> 主题:Re: 基于flink 1.12 如何通过Flink SQL 实现 多sink端 在同一个事务中
>
>
>
> Hi！
>
> 如果 statement set 已经包含了多个 insert 语句，那么写入 kafka 表和写入 db 应该是在同一个作业里进行的。如果写入 db
> 失败，那么产生的 exception 应该会让作业失败才对。这里 db 写入失败但 kafka 依旧写入是什么样的现象？
>
> 另外其实可以考虑分成两个作业，第一个作业将数据写入 db，第二个作业从 db 读出数据写入 kafka。关于捕获 db 数据的变化，可以看一下
> Flink CDC connector[1]
>
> [1] https://github.com/ververica/flink-cdc-connectors
>
> 悟空 
>  能否详细说下呢，statement set[1] 是什么意思， 我是通过StatementSet.addInsertSql 将多个sql
>  加入的，然后执行execute()方法
> 
> 
> 
> 
>  --nbsp;原始邮件nbsp;--
>  发件人:
> 
> "user-zh"
> 
> <
>  fskm...@gmail.comgt;;
>  发送时间:nbsp;2021年8月26日(星期四) 中午12:36
>  收件人:nbsp;"user-zh" 
>  主题:nbsp;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
> 
> 
> ;
>  悟空  
>  gt; hi all:amp;nbsp;
>  gt; amp;nbsp; amp;nbsp; 我目前基于flink 1.12 sql 来开发功能，
> 目前遇到一个问题， 我现在想实现
>  在一个事务里 先将kafka
>  gt; 源的数据写入到一张msyql 表中， 写入成功后再写入一张kafka的表里。如果写入db失败，则不写入kafka 。
>  gt; amp;nbsp; amp;nbsp;语句类似这种：
>  gt; amp;nbsp; amp;nbsp;insert into
> db_table_sinkamp;nbsp;select *
>  fromamp;nbsp;
>  gt; kafka_source_table;
>  gt; amp;nbsp; amp;nbsp;insert into kafka_table_sink
> select * from
>  kafka_source_table;
>  gt;
>  gt;
>  gt; amp;nbsp; 请问flink SQL 有实现方式吗？
> 目前经过测试，发现如果db写入失败，但是kafka依旧写入，同时flink
>  程序没有挂掉。

flink run 提交任务到yarn 报Cannot load user class: org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

2021-08-30 文章 Wayne

我的执行命令是
 flink run -m yarn-cluster -yd -yjm 1024m -ytm 1024m -ys 1 -ynm xxx -C 
file:///xxx/flink-connector-kafka_2.12-1.12.2.jar -C 
file:///xxx/flink-sql-avro-1.12.2.jar ...
提交到生产集群上提示 
org.apache.flink.streaming.runtime.tasks.StreamTaskException: Cannot load user 
class: org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
ClassLoader info: URL ClassLoader:
Class not resolvable through given classloader.
at 
org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:331)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.streaming.runtime.tasks.OperatorChain.(OperatorChain.java:145)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:517)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:573) 
~[flink-dist_2.12-1.12.2.jar:1.12.2]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) 
~[flink-dist_2.12-1.12.2.jar:1.12.2]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) 
~[flink-dist_2.12-1.12.2.jar:1.12.2]
at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_292]
Caused by: java.lang.ClassNotFoundException: 
org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
at java.net.URLClassLoader.findClass(URLClassLoader.java:382) 
~[?:1.8.0_292]
at java.lang.ClassLoader.loadClass(ClassLoader.java:418) ~[?:1.8.0_292]
at 
org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:64)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at java.lang.ClassLoader.loadClass(ClassLoader.java:351) ~[?:1.8.0_292]
at 
org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoaders$SafetyNetWrapperClassLoader.loadClass(FlinkUserCodeClassLoaders.java:172)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at java.lang.Class.forName0(Native Method) ~[?:1.8.0_292]
at java.lang.Class.forName(Class.java:348) ~[?:1.8.0_292]
at 
org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:76)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1986) 
~[?:1.8.0_292]
at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1850) 
~[?:1.8.0_292]
at 
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2160) 
~[?:1.8.0_292]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667) 
~[?:1.8.0_292]
at 
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405) 
~[?:1.8.0_292]
at 
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329) 
~[?:1.8.0_292]
at 
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187) 
~[?:1.8.0_292]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667) 
~[?:1.8.0_292]
at 
java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405) 
~[?:1.8.0_292]
at 
java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329) 
~[?:1.8.0_292]
at 
java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187) 
~[?:1.8.0_292]
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667) 
~[?:1.8.0_292]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:503) 
~[?:1.8.0_292]
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:461) 
~[?:1.8.0_292]
at 
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:600)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:587)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:541)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
at 
org.apache.flink.streaming.api.graph.StreamConfig.getStreamOperatorFactory(StreamConfig.java:317)
 ~[flink-dist_2.12-1.12.2.jar:1.12.2]
... 6 more
2


任务可以提交到yarn上，说明 -C 这个命令起作用了，但是为啥yarn上报错,是我这个命令写的不对么

退订

2021-08-30 文章罗海芳





| |
罗海芳
|
|
15678617...@163.com
|
签名由网易邮箱大师定制

Flink on yarn的日志监控和checkpoint的监控生产是如何处理的？

Re: flink run -d -m yarn-cluster 提交任务到yarn集群不成功

Re: flink run -d -m yarn-cluster 提交任务到yarn集群不成功

Re: flink oss ha

Re: flink run 提交任务到yarn 报Cannot load user class: org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

退订

??????flink keyby????

flink keyby????

Re: Flink 从checkpoint恢复时，部分状态没有正确恢复

Re: Flink 从checkpoint恢复时，部分状态没有正确恢复

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

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

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

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

flink run 提交任务到yarn 报Cannot load user class: org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

退订

16 matches

Site Navigation

Mail list logo

Footer information