flink-sql-gateway相关问题
请问有关于flink-sql-gateway rest api以pre job模式提交作业到yarn集群的文档吗？
python udf求助: Process died with exit code 0
我进行了如下操作： https://yuchi.blog.csdn.net/article/details/112837327 然后报错: java.lang.IllegalStateException: Process died with exit code 0 请问应该如何解决？ 谢谢您～！
Re: 回复：请教关于Flink yarnship的使用
我试过了，还是null -- Sent from: http://apache-flink.147419.n8.nabble.com/
flink启动任务时异常，加载不到flink/lib目录下的包
环境： flink 1.12.0 报这个错：报加载不到这个log4j-slf4j-impl-2.12.1.jar包，但是我的lib目录下是有这个包的： $ cd lib/ $ ll -rw-r--r-- 1 yujianbo yujianbo 91554 1月 26 18:44 flink-csv-1.12.0.jar -rw-r--r-- 1 yujianbo yujianbo 114119885 1月 26 18:45 flink-dist_2.11-1.12.0.jar -rw-r--r-- 1 yujianbo yujianbo136664 1月 26 18:44 flink-json-1.12.0.jar -rw-r--r-- 1 yujianbo yujianbo 7709742 1月 26 18:44 flink-shaded-zookeeper-3.4.14.jar -rw-r--r-- 1 yujianbo yujianbo 36147824 1月 26 18:44 flink-table_2.11-1.12.0.jar -rw-r--r-- 1 yujianbo yujianbo 40286363 1月 26 18:45 flink-table-blink_2.11-1.12.0.jar -rw-r--r-- 1 yujianbo yujianbo 67114 1月 26 18:44 log4j-1.2-api-2.12.1.jar -rw-r--r-- 1 yujianbo yujianbo276771 1月 26 18:44 log4j-api-2.12.1.jar -rw-r--r-- 1 yujianbo yujianbo 1674433 1月 26 18:44 log4j-core-2.12.1.jar -rw-r--r-- 1 yujianbo yujianbo 23518 1月 26 18:44 log4j-slf4j-impl-2.12.1.jar The program finished with the following exception: org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: Could not deploy Yarn job cluster. at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:330) at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:198) at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114) at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:743) at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:242) at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:971) at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1047) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subject.doAs(Subject.java:422) at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1845) at org.apache.flink.runtime.security.contexts.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1047) Caused by: org.apache.flink.client.deployment.ClusterDeploymentException: Could not deploy Yarn job cluster. at org.apache.flink.yarn.YarnClusterDescriptor.deployJobCluster(YarnClusterDescriptor.java:460) at org.apache.flink.client.deployment.executors.AbstractJobClusterExecutor.execute(AbstractJobClusterExecutor.java:70) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1940) at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:128) at org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76) at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1822) at com.xm4399.yhzx.task.VersionTest.main(VersionTest.java:118) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:316) ... 11 more Caused by: org.apache.flink.yarn.YarnClusterDescriptor$YarnDeploymentException: The YARN application unexpectedly switched to state FAILED during deployment. Diagnostics from YARN: Application application_1611303948765_0050 failed 1 times (global limit =2; local limit is =1) due to AM Container for appattempt_1611303948765_0050_01 exited with exitCode: -1000 *Failing this attempt.Diagnostics: [2021-01-26 19:00:46.608]File does not exist: hdfs://4399cluster/user/hadoop/.flink/application_1611303948765_0050/lib/log4j-slf4j-impl-2.12.1.jar java.io.FileNotFoundException: File does not exist: hdfs://4399cluster/user/hadoop/.flink/application_1611303948765_0050/lib/log4j-slf4j-impl-2.12.1.jar* at org.apache.hadoop.hdfs.DistributedFileSystem$29.doCall(DistributedFileSystem.java:1729) at org.apache.hadoop.hdfs.DistributedFileSystem$29.doCall(DistributedFileSystem.java:1722) at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81) at org.apache.hadoop.hdfs.DistributedFileSystem.getFileStatus(DistributedFileSystem.java:1737) at org.apache.hadoop.yarn.util.FSDownload.verifyAndCopy(FSDownload.java:271) at org.apache.hadoop.yarn.util.FSDownload.access$000(FSDownload.java:68) at org.apache.hadoop.yarn.util.FSDownload$2.run(FSDownload.java:415) at org.apache.hadoop.yarn.util.FSDownload$2.run(FSDownload.java:412) at java.security.AccessController.doPrivileged(Native Method) at javax.security.auth.Subj
[DISCUSS] Removal of flink-swift-fs-hadoop module
Hi all, during a security maintenance PR [1], Chesnay noticed that the flink-swift-fs-hadoop module is lacking test coverage [2]. Also, there hasn't been any substantial change since 2018, when it was introduced. On the user@ ML, I could not find any proof of significant use of the module (no one mentioned any problems with it). *I propose to remove this module in Flink 1.13*. Otherwise, we would release a module with known vulnerable dependencies, and an unknown stability. If there are users, they can still use the 1.12.0 release of it. If we notice that there are a lot of users, we can reintroduce the FS, and add proper tests for it. Please let me know if you have any concerns, otherwise, I'll remove it. [1] https://github.com/apache/flink/pull/14749 [2] https://issues.apache.org/jira/browse/FLINK-20804
Re: flink-sql-gateway支持远程吗
flink-sql-gateway在提交job时，会根据SPI加载executor，然后推断是local mode还是remote mode, 在remote mode下，根据FLINK_HOME 目录下config.sh推断的FLINK_CONF_DIR 寻找flink-conf.yaml， 其中的host, rest port决定了提交的远端集群 罗显宴 <15927482...@163.com> 于2021年1月23日周六 上午11:19写道： > > 大佬们，我试过flinksql-client和flink-sql-gateway发现都是把代码提交到本地集群运行，好像不能提交到指定远程集群上执行任务，请问大佬们，怎么提交代码到远程集群运行 > > > | | > 15927482803 > | > | > 邮箱：15927482...@163.com > | > > 签名由 网易邮箱大师 定制 -- *With kind regards Sebastian Liu 刘洋 Institute of Computing Technology, Chinese Academy of Science Mobile\WeChat: +86—15201613655 E-mail: liuyang0...@gmail.com QQ: 3239559*
Re: flink-sql-gateway相关问题
sql gateway 提交job方式和flink client类似主要取决于flink-conf中的execution.config 配置, 对per job模式on yarn, 对应的配置是“yarn-per-job”, 这样加载的PipelineExecutor会控制为：YarnJobClusterExecutorFactory，具备向Yarn提交job的能力，这和使用flink client 提交job最终是一个调用路径，余下在flink-conf中添加上yarn相关配置我理解就可以了。 org.apache.flink.yarn.configuration.YarnConfigOptions zilong xiao 于2021年1月26日周二 下午4:00写道： > 请问有关于flink-sql-gateway rest api以pre job模式提交作业到yarn集群的文档吗？ > -- *With kind regards Sebastian Liu 刘洋 Institute of Computing Technology, Chinese Academy of Science Mobile\WeChat: +86—15201613655 E-mail: liuyang0...@gmail.com QQ: 3239559*
Re: Caused by: java.lang.NoSuchMethodError: com.google.common.base.Preconditions.checkArgument(ZLjava/lang/String;Ljava/lang/Object;)V
Rui Li 上午好，能帮我看一下这个http://apache-flink.147419.n8.nabble.com/flink-flink-lib-td10518.html也是由依赖冲突引起的吗。我的情况是这样的，一样的集群，之前是cdh的hadoop3.0.0，hive是2.2.0；现在准备从cdh迁回社区版的hadoop集群，变成3.3.0，hive是3.1.2，昨天解决了hive-exec的问题，但是今天同样的一份代码在上一个集群是可以正常提交正常跑的，但是放到新的集群这边跑，启动的时候就直接说加载不到lib包下的一个jar，就有点奇怪，是yarn冲突了吗？ -- Sent from: http://apache-flink.147419.n8.nabble.com/
Re: flink启动任务时异常，加载不到flink/lib目录下的包
我已经解决了，是因为我跑的任务jar包里的resources目录下的一个文件hdfs-site.xml是上一份集群的，我把这个去掉就可以了。 -- Sent from: http://apache-flink.147419.n8.nabble.com/
Re: flink-sql-gateway相关问题
感谢您的回答，我在flink-conf.yaml里指定"execution.target: yarn-pre-job"后，尝试用rest
api生成session id时会遇到异常，不清楚是为何，可否帮忙看下
flink version: 1.11.3
execution.target: yarn-pre-job
rest api请求路径和参数：
http://localhost:8083/v1/sessions
{
"planner": "blink",
"execution_type": "streaming"
}
异常信息：Caused by: java.lang.IllegalStateException: No ClusterClientFactory
found. If you were targeting a Yarn cluster, please make sure to export the
HADOOP_CLASSPATH environment variable or have hadoop in your classpath. For
more information refer to the "Deployment & Operations" section of the
official Apache Flink documentation.
Sebastian Liu 于2021年1月27日周三 上午1:01写道：
> sql gateway 提交job方式和flink client类似主要取决于flink-conf中的execution.config 配置,
> 对per job模式on yarn, 对应的配置是“yarn-per-job”,
>
> 这样加载的PipelineExecutor会控制为：YarnJobClusterExecutorFactory，具备向Yarn提交job的能力，这和使用flink
> client
> 提交job最终是一个调用路径，余下在flink-conf中添加上yarn相关配置我理解就可以了。
> org.apache.flink.yarn.configuration.YarnConfigOptions
>
> zilong xiao 于2021年1月26日周二 下午4:00写道：
>
> > 请问有关于flink-sql-gateway rest api以pre job模式提交作业到yarn集群的文档吗？
> >
>
>
> --
>
> *With kind regards
>
> Sebastian Liu 刘洋
> Institute of Computing Technology, Chinese Academy of Science
> Mobile\WeChat: +86—15201613655
> E-mail: liuyang0...@gmail.com
> QQ: 3239559*
>
Re: flink-sql-gateway支持远程吗
如果是yarn-pre-job，是如何提交到yarn集群的呢，跟host * rest port应该没关系了吧？ Sebastian Liu 于2021年1月27日周三 上午12:48写道： > flink-sql-gateway在提交job时，会根据SPI加载executor，然后推断是local mode还是remote mode, > 在remote mode下，根据FLINK_HOME 目录下config.sh推断的FLINK_CONF_DIR 寻找flink-conf.yaml， > 其中的host, rest port决定了提交的远端集群 > > 罗显宴 <15927482...@163.com> 于2021年1月23日周六 上午11:19写道： > > > > > > 大佬们，我试过flinksql-client和flink-sql-gateway发现都是把代码提交到本地集群运行，好像不能提交到指定远程集群上执行任务，请问大佬们，怎么提交代码到远程集群运行 > > > > > > | | > > 15927482803 > > | > > | > > 邮箱：15927482...@163.com > > | > > > > 签名由 网易邮箱大师 定制 > > > > -- > > *With kind regards > > Sebastian Liu 刘洋 > Institute of Computing Technology, Chinese Academy of Science > Mobile\WeChat: +86—15201613655 > E-mail: liuyang0...@gmail.com > QQ: 3239559* >
Re: flink-sql-gateway支持远程吗
Hi zilong, sql gateway复用了很多底层API，其提交方式很多复用了flink client的逻辑。 针对提交模式，主要是看flink-conf.yaml中的的execution.config 配置, 关于这个配置 可以参考：org.apache.flink.configuration.DeploymentOptions#TARGET 如果是yarn-per-job, 需要配置为：“yarn-per-job”, 同时也依赖flink-conf.yaml中其他yarn的相关配置： 可参考：org.apache.flink.yarn.configuration.YarnConfigOptions 总结一下就是： 1. sql gateway找到正确的flink-conf.yaml, 取决于FLINK_HOME or FLINK_CONF_DIR. 2. 根据execution.config选择对应executor，然后不同模式依赖于flink-conf对应的提交相关配置 zilong xiao 于2021年1月27日周三 上午10:19写道： > 如果是yarn-pre-job，是如何提交到yarn集群的呢，跟host * rest port应该没关系了吧？ > > Sebastian Liu 于2021年1月27日周三 上午12:48写道： > > > flink-sql-gateway在提交job时，会根据SPI加载executor，然后推断是local mode还是remote mode, > > 在remote mode下，根据FLINK_HOME 目录下config.sh推断的FLINK_CONF_DIR > 寻找flink-conf.yaml， > > 其中的host, rest port决定了提交的远端集群 > > > > 罗显宴 <15927482...@163.com> 于2021年1月23日周六 上午11:19写道： > > > > > > > > > > > 大佬们，我试过flinksql-client和flink-sql-gateway发现都是把代码提交到本地集群运行，好像不能提交到指定远程集群上执行任务，请问大佬们，怎么提交代码到远程集群运行 > > > > > > > > > | | > > > 15927482803 > > > | > > > | > > > 邮箱：15927482...@163.com > > > | > > > > > > 签名由 网易邮箱大师 定制 > > > > > > > > -- > > > > *With kind regards > > > > Sebastian Liu 刘洋 > > Institute of Computing Technology, Chinese Academy of Science > > Mobile\WeChat: +86—15201613655 > > E-mail: liuyang0...@gmail.com > > QQ: 3239559* > > > -- *With kind regards Sebastian Liu 刘洋 Institute of Computing Technology, Chinese Academy of Science Mobile\WeChat: +86—15201613655 E-mail: liuyang0...@gmail.com QQ: 3239559*
Re: flink-sql-gateway相关问题
看报错提示，应该是没有指定正确HADOOP_HOME env? 如果用flink client提交，也是需要这个env的
zilong xiao 于2021年1月27日周三 上午10:17写道：
> 感谢您的回答，我在flink-conf.yaml里指定"execution.target: yarn-pre-job"后，尝试用rest
> api生成session id时会遇到异常，不清楚是为何，可否帮忙看下
>
> flink version: 1.11.3
> execution.target: yarn-pre-job
> rest api请求路径和参数：
> http://localhost:8083/v1/sessions
> {
> "planner": "blink",
> "execution_type": "streaming"
>}
>
> 异常信息：Caused by: java.lang.IllegalStateException: No ClusterClientFactory
> found. If you were targeting a Yarn cluster, please make sure to export the
> HADOOP_CLASSPATH environment variable or have hadoop in your classpath. For
> more information refer to the "Deployment & Operations" section of the
> official Apache Flink documentation.
>
> Sebastian Liu 于2021年1月27日周三 上午1:01写道：
>
> > sql gateway 提交job方式和flink client类似主要取决于flink-conf中的execution.config 配置,
> > 对per job模式on yarn, 对应的配置是“yarn-per-job”,
> >
> >
> 这样加载的PipelineExecutor会控制为：YarnJobClusterExecutorFactory，具备向Yarn提交job的能力，这和使用flink
> > client
> > 提交job最终是一个调用路径，余下在flink-conf中添加上yarn相关配置我理解就可以了。
> > org.apache.flink.yarn.configuration.YarnConfigOptions
> >
> > zilong xiao 于2021年1月26日周二 下午4:00写道：
> >
> > > 请问有关于flink-sql-gateway rest api以pre job模式提交作业到yarn集群的文档吗？
> > >
> >
> >
> > --
> >
> > *With kind regards
> >
> > Sebastian Liu 刘洋
> > Institute of Computing Technology, Chinese Academy of Science
> > Mobile\WeChat: +86—15201613655
> > E-mail: liuyang0...@gmail.com
> > QQ: 3239559*
> >
>
--
*With kind regards
Sebastian Liu 刘洋
Institute of Computing Technology, Chinese Academy of Science
Mobile\WeChat: +86—15201613655
E-mail: liuyang0...@gmail.com
QQ: 3239559*
TaggedOperatorSubtaskState is missing when creating a new savepoint using state processor api
Hi, 各位好：
我在使用state processor api创建新的包含kafka相关state的savepoint用来修改max
parallelism时，创建成功后使用此savepoint来重启任务，发现抛出如下异常：
{code}
java.lang.Exception: Exception while creating StreamOperatorStateContext.
at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:204)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:247)
at
org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:290)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$0(StreamTask.java:473)
at
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$SynchronizedStreamTaskActionExecutor.runThrowing(StreamTaskActionExecutor.java:92)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:469)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:522)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:721)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.util.FlinkException: Could not restore operator
state backend for StreamSource_e8ea6e352a1a627513ffbd4573fa1628_(1/1) from any
of the 1 provided restore options.
at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)
at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.operatorStateBackend(StreamTaskStateInitializerImpl.java:265)
at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:152)
... 9 more
Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when
trying to restore operator state backend
at
org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder.build(DefaultOperatorStateBackendBuilder.java:86)
at
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createOperatorStateBackend(RocksDBStateBackend.java:552)
at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$operatorStateBackend$0(StreamTaskStateInitializerImpl.java:256)
at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142)
at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121)
... 11 more
Caused by: java.lang.IllegalStateException: Missing value for the key
'org.apache.flink.state.api.output.TaggedOperatorSubtaskState'
at
org.apache.flink.util.LinkedOptionalMap.unwrapOptionals(LinkedOptionalMap.java:190)
at
org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializerSnapshot.restoreSerializer(KryoSerializerSnapshot.java:86)
at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
at
java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
at
java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545)
at
java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)
at
org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(NestedSerializersSnapshotDelegate.java:225)
at
org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(NestedSerializersSnapshotDelegate.java:83)
at
org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:204)
at
org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:189)
at
org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:164)
at
org.apache.flink.runtime.state.RegisteredOperatorStateBackendMetaInfo.getPartitionStateSerializer(RegisteredOperatorStateBackendMetaInfo.java:113)
at
org.apache.flink.runtime.state.OperatorStateRestoreOperation.restore(OperatorStateRestoreOperation.java:94)
at
org.apache.flink.runtime.state.DefaultOperatorStateBackendBuilder.build(DefaultOperatorStateBackendBuilder.java:83)
... 15 more
{code}
我的使用方式如下：
@Override
public void createNewSavepoint(ExecutionEnvironment env, String savepointPath,
StateBackend stateBackend,
ParameterTool config) {
// 加载未修改max parallelism的savepoint
String savepointOutputPath =
config.get(EapSavepointConstants.EAP_SAVEPOINT_OUTPUT_PATH);
// 新的max parallelism
int maxP
Flink sql 1.12写入hive报metastore失败
使用flink sql 1.12写入hive,未提交到yarn上成功，错误信息如下： 2021-01-26 20:44:23.133 [main] INFO org.apache.hadoop.hive.metastore.HiveMetaStoreClient - Trying to connect to metastore with URI thrift://hdcom02.prd.com:9083 2021-01-26 20:44:23.133 [main] INFO org.apache.hadoop.hive.metastore.HiveMetaStoreClient - Trying to connect to metastore with URI thrift://hdcom02.prd.com:9083 2021-01-26 20:44:23.134 [main] INFO org.apache.hadoop.hive.metastore.HiveMetaStoreClient - Opened a connection to metastore, current connections: 2 2021-01-26 20:44:23.134 [main] INFO org.apache.hadoop.hive.metastore.HiveMetaStoreClient - Opened a connection to metastore, current connections: 2 2021-01-26 20:44:23.181 [main] WARN org.apache.hadoop.hive.metastore.HiveMetaStoreClient - set_ugi() not successful, Likely cause: new client talking to old server. Continuing without it. org.apache.thrift.transport.TTransportException: null at org.apache.thrift.transport.TIOStreamTransport.read(TIOStreamTransport.java:132) at org.apache.thrift.transport.TTransport.readAll(TTransport.java:86) at org.apache.thrift.protocol.TBinaryProtocol.readStringBody(TBinaryProtocol.java:380) at org.apache.thrift.protocol.TBinaryProtocol.readMessageBegin(TBinaryProtocol.java:230) at org.apache.thrift.TServiceClient.receiveBase(TServiceClient.java:77) at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$Client.recv_set_ugi(ThriftHiveMetastore.java:4787) at org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore$Client.set_ugi(ThriftHiveMetastore.java:4773) at org.apache.hadoop.hive.metastore.HiveMetaStoreClient.open(HiveMetaStoreClient.java:534) at org.apache.hadoop.hive.metastore.HiveMetaStoreClient.(HiveMetaStoreClient.java:224) at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) at java.lang.reflect.Constructor.newInstance(Constructor.java:423) at org.apache.hadoop.hive.metastore.utils.JavaUtils.newInstance(JavaUtils.java:84) at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.(RetryingMetaStoreClient.java:95) at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.getProxy(RetryingMetaStoreClient.java:148) at org.apache.hadoop.hive.metastore.RetryingMetaStoreClient.getProxy(RetryingMetaStoreClient.java:104) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) at org.apache.flink.table.catalog.hive.client.HiveShimV310.getHiveMetastoreClient(HiveShimV310.java:112) at org.apache.flink.table.catalog.hive.client.HiveMetastoreClientWrapper.createMetastoreClient(HiveMetastoreClientWrapper.java:274) at org.apache.flink.table.catalog.hive.client.HiveMetastoreClientWrapper.(HiveMetastoreClientWrapper.java:80) at org.apache.flink.table.catalog.hive.client.HiveMetastoreClientFactory.create(HiveMetastoreClientFactory.java:32) at org.apache.flink.connectors.hive.HiveTableSink.consume(HiveTableSink.java:145) at org.apache.flink.connectors.hive.HiveTableSink.lambda$getSinkRuntimeProvider$0(HiveTableSink.java:137) at org.apache.flink.table.planner.plan.nodes.common.CommonPhysicalSink.createSinkTransformation(CommonPhysicalSink.scala:95) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.scala:109) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.scala:43) at org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan(ExecNode.scala:59) at org.apache.flink.table.planner.plan.nodes.exec.ExecNode.translateToPlan$(ExecNode.scala:57) at org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlan(StreamExecSink.scala:43) at org.apache.flink.table.planner.delegation.StreamPlanner.$anonfun$translateToPlan$1(StreamPlanner.scala:66) at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233) at scala.collection.Iterator.foreach(Iterator.scala:937) at scala.collection.Iterator.foreach$(Iterator.scala:937) at scala.collection.AbstractIterator.foreach(Iterator.scala:1425) at scala.collection.IterableLike.foreach(IterableLike.scala:70) at scala.collection.IterableLike.foreach$(IterableLike.scala:69) at scala.collection.AbstractIterable.foreach(Iterable.scala:54) at scala.collection.TraversableLike.map(Traversabl
Flink 1.10 table setIdleStateRetentionTime
hi, Flink 1.10.1 on yarn，测试发现，`tableEnv.getConfig.setIdleStateRetentionTime(Time.days(180), Time.days(181))` 修改为 `tableEnv.getConfig.setIdleStateRetentionTime(Time.days(30), Time.days(31))`。在savePoInt中恢复 job 启动正常，运行的时候 在webUI Exception 中有连接 kafka 异常，Timeout expired while fetching topic metadata。不太明白是什么原因产生这个？是由于maxTime 问题吗？
Re: Caused by: java.lang.NoSuchMethodError: com.google.common.base.Preconditions.checkArgument(ZLjava/lang/String;Ljava/lang/Object;)V
参考大佬Rui Li的建议，我解决了，想参考的可以看看这个： https://blog.csdn.net/weixin_44500374/article/details/113244560 https://www.jianshu.com/p/f076a4f66527 -- Sent from: http://apache-flink.147419.n8.nabble.com/
Re: Caused by: java.lang.NoSuchMethodError: com.google.common.base.Preconditions.checkArgument(ZLjava/lang/String;Ljava/lang/Object;)V
我的已经解决了，根据大佬Rui Li的建议，也可以参考我的方式： https://blog.csdn.net/weixin_44500374/article/details/113244560 https://www.jianshu.com/p/f076a4f66527 -- Sent from: http://apache-flink.147419.n8.nabble.com/
Re: Flink 1.10 table setIdleStateRetentionTime
抱歉..弄错了 On 1/27/2021 11:39，xuhaiLong wrote： hi, Flink 1.10.1 on yarn，测试发现，`tableEnv.getConfig.setIdleStateRetentionTime(Time.days(180), Time.days(181))` 修改为 `tableEnv.getConfig.setIdleStateRetentionTime(Time.days(30), Time.days(31))`。在savePoInt中恢复 job 启动正常，运行的时候 在webUI Exception 中有连接 kafka 异常，Timeout expired while fetching topic metadata。不太明白是什么原因产生这个？是由于maxTime 问题吗？
用application mode部署应用，classloader.resolve-order参数是否必须要改为parent-first？
我写了一个 process function的demo，自定义source产生数据sink到kafka，然后发布到yarn集群运行，flink版本是1.11.2，采用application Mode 部署，然后发现jobmanager-log报错： Failed to construct kafka producer；Caused by: org.apache.kafka.common.KafkaException: class org.apache.kafka.common.serialization.ByteArraySerializer is not an instance of org.apache.kafka.common.serialization.Serializer。 换了flink版本为1.12.1发现还是报这个错，后尝试采用per-job Mode部署发现是OK的。查资料发现是跟flink的类加载方式有关，即flink-conf.yml中的classloader.resolve-order参数，要将默认的 child-first改成parent-first，修改后确实ok了，但是有个疑惑，为啥要改这个参数呢，看了官方文档，一般不建议改这个参数的，他避免了使用flink内置的类加载器，而是使用APP自己的。 -- Sent from: http://apache-flink.147419.n8.nabble.com/
Re: flink-sql-gateway稳定性如何，可以在生产环境使用吗？
如果使用zeppelin，zeppelin可以提供rpc接口吗 -- Sent from: http://apache-flink.147419.n8.nabble.com/
Scala REPL YARN 运行模式报 NoSuchMethodError setPrintSpaceAfterFullCompletion
bin/start-scala-shell.sh yarn scala> Exception in thread "main" java.lang.NoSuchMethodError: jline.console.completer.CandidateListCompletionHandler.setPrintSpaceAfterFullCompletion(Z)V at scala.tools.nsc.interpreter.jline.JLineConsoleReader.initCompletion(JLineReader.scala:139) at scala.tools.nsc.interpreter.jline.InteractiveReader.postInit(JLineReader.scala:54) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1$$anonfun$25.apply(ILoop.scala:899) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1$$anonfun$25.apply(ILoop.scala:897) at scala.tools.nsc.interpreter.SplashReader.postInit(InteractiveReader.scala:130) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1$$anonfun$scala$tools$nsc$interpreter$ILoop$$anonfun$$loopPostInit$1$1.apply$mcV$sp(ILoop.scala:926) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1$$anonfun$scala$tools$nsc$interpreter$ILoop$$anonfun$$loopPostInit$1$1.apply(ILoop.scala:908) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1$$anonfun$scala$tools$nsc$interpreter$ILoop$$anonfun$$loopPostInit$1$1.apply(ILoop.scala:908) at scala.tools.nsc.interpreter.ILoop$$anonfun$mumly$1.apply(ILoop.scala:189) at scala.tools.nsc.interpreter.IMain.beQuietDuring(IMain.scala:221) at scala.tools.nsc.interpreter.ILoop.mumly(ILoop.scala:186) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1$$anonfun$startup$1$1.apply(ILoop.scala:979) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply$mcZ$sp(ILoop.scala:990) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:891) at scala.tools.nsc.interpreter.ILoop$$anonfun$process$1.apply(ILoop.scala:891) at scala.reflect.internal.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:97) at scala.tools.nsc.interpreter.ILoop.process(ILoop.scala:891) at org.apache.flink.api.scala.FlinkShell$.startShell(FlinkShell.scala:184) at org.apache.flink.api.scala.FlinkShell$.main(FlinkShell.scala:131) at org.apache.flink.api.scala.FlinkShell.main(FlinkShell.scala) va Exception in thread "Thread-2" java.lang.InterruptedException at java.util.concurrent.SynchronousQueue.put(SynchronousQueue.java:879) at scala.tools.nsc.interpreter.SplashLoop.run(InteractiveReader.scala:77) at java.lang.Thread.run(Thread.java:748) ```
Re: flink-sql-gateway稳定性如何，可以在生产环境使用吗？
zeppelin 有 rest api 接口，https://www.yuque.com/jeffzhangjianfeng/gldg8w/pz2xoh jinsx 于2021年1月27日周三 下午2:30写道： > 如果使用zeppelin，zeppelin可以提供rpc接口吗 > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/ > -- Best Regards Jeff Zhang
Re: 用application mode部署应用，classloader.resolve-order参数是否必须要改为parent-first？
你是打包的时候自己打了 kafka 的依赖进去吗？看起来是应用里有一个 kafka 版本 A，接口 org.apache.kafka.common.serialization.Serializer 用应用 classloader 加载，然后 flink kafka connector 用集群 classloader 加载，继承自集群 classloader 里的 org.apache.kafka.common.serialization.Serializer 导致这个问题。 Best, tison. lp <973182...@qq.com> 于2021年1月27日周三 下午12:39写道： > 我写了一个 process > > function的demo，自定义source产生数据sink到kafka，然后发布到yarn集群运行，flink版本是1.11.2，采用application > Mode 部署，然后发现jobmanager-log报错： Failed to construct kafka producer；Caused by: > org.apache.kafka.common.KafkaException: class > org.apache.kafka.common.serialization.ByteArraySerializer is not an > instance > of org.apache.kafka.common.serialization.Serializer。 > 换了flink版本为1.12.1发现还是报这个错，后尝试采用per-job > > Mode部署发现是OK的。查资料发现是跟flink的类加载方式有关，即flink-conf.yml中的classloader.resolve-order参数，要将默认的 > > child-first改成parent-first，修改后确实ok了，但是有个疑惑，为啥要改这个参数呢，看了官方文档，一般不建议改这个参数的，他避免了使用flink内置的类加载器，而是使用APP自己的。 > > > > -- > Sent from: http://apache-flink.147419.n8.nabble.com/
Re: 用application mode部署应用，classloader.resolve-order参数是否必须要改为parent-first？
又或者是反过来。 你可以尝试把 kafka connector 放到 /libs 里，自己应用打包里不带 kafka jar 也不带 kafka connector jar，应该就可以不改配置。 Best, tison. tison 于2021年1月27日周三 下午2:47写道： > 你是打包的时候自己打了 kafka 的依赖进去吗？看起来是应用里有一个 kafka 版本 A，接口 > org.apache.kafka.common.serialization.Serializer 用应用 classloader 加载，然后 > flink kafka connector 用集群 classloader 加载，继承自集群 classloader 里的 > org.apache.kafka.common.serialization.Serializer 导致这个问题。 > > Best, > tison. > > > lp <973182...@qq.com> 于2021年1月27日周三 下午12:39写道： > >> 我写了一个 process >> >> function的demo，自定义source产生数据sink到kafka，然后发布到yarn集群运行，flink版本是1.11.2，采用application >> Mode 部署，然后发现jobmanager-log报错： Failed to construct kafka producer；Caused >> by: >> org.apache.kafka.common.KafkaException: class >> org.apache.kafka.common.serialization.ByteArraySerializer is not an >> instance >> of org.apache.kafka.common.serialization.Serializer。 >> 换了flink版本为1.12.1发现还是报这个错，后尝试采用per-job >> >> Mode部署发现是OK的。查资料发现是跟flink的类加载方式有关，即flink-conf.yml中的classloader.resolve-order参数，要将默认的 >> >> child-first改成parent-first，修改后确实ok了，但是有个疑惑，为啥要改这个参数呢，看了官方文档，一般不建议改这个参数的，他避免了使用flink内置的类加载器，而是使用APP自己的。 >> >> >> >> -- >> Sent from: http://apache-flink.147419.n8.nabble.com/ > >
flink-sql-gateway如何使用flink自定义的udf
各位大佬，Flink-sql-gateway 提交flink sql任务 ，如何使用flink自定义的udf | | 阿华田 | | a15733178...@163.com | 签名由网易邮箱大师定制
pyflink 1.11 connector数据读取问题
目前通过 connector 定义ddl的方式，通过数据库读取数据，方式如下： CREATE TABLE MyUserTable ( id BIGINT, name STRING, age INT, status BOOLEAN, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/mydatabase', 'table-name' = 'users' ); 每次表格获取及定义都需要定义数据类型（如例子中: BIGINT,STRING,INT,BOOLEN）， 请问大佬是否有其他从数据库读取方式，可以不用定义数据类型??
回复：pyflink 1.11 connector数据读取问题
写个程序自动根据表的元数据生成ddl语句 ---原始邮件--- 发件人: "肖越"<18242988...@163.com> 发送时间: 2021年1月27日(周三) 下午3:29 收件人: "user-zh"
