Flink报错问题之写orc报错如何解决

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: Apache Flink是由Apache软件基金会开发的开源流处理框架,其核心是用Java和Scala编写的分布式流数据流引擎。本合集提供有关Apache Flink相关技术、使用技巧和最佳实践的资源。

问题一:flink使用多个keytab的情况怎么处理?

你好,我在使用flink run提交任务时,由于集群是kerberos化的Hadoop集群,我想为不同的用户提供不同的keytab进行认证,但是我在flink-conf.yaml中看到了的security.kerberos.login.keytab和security.kerberos.login.principal的相关配置,但是这不能动态配置;我尝试了使用 -yD进行配置,但是并不起作用,我想知道如何处理这样的情况?期待您的回复~~~ *来自志愿者整理的flink邮件归档



参考答案:

可以在调用executeSql()或execute()(多SQL)的时候使用ugi认证进行任务提交,以及在run命令中动态传入你所说的两个参数,以-yD的形式。*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370072?spm=a2c6h.13066369.question.36.33bf585fTaxMq8



问题二:flink 1.11.2写hive 2.1.1 orc 遇到的问题

flink 版本1.11.2 hive 版本2.1.1 基于cdh 6.2.1 写普通表或parquet没问题,写orc报如下错误: http://apache-flink.147419.n8.nabble.com/file/t1150/flink_hive%E5%8C%85%E5%86%B2%E7%AA%81.png

也看到其它邮件列表说修改: flink-sql-connector-hive-2.2.0_2.11-1.11.2.jar OrcFile: WriterVersion CURRENT_WRITER = WriterVersion.HIVE_13083 重新编译即可,但是这样尝试之后还是报同样的错误,是Hive必须升级到3.x版本么?*来自志愿者整理的flink邮件归档



参考答案:

你的flink lib下都添加了哪些依赖呢,另外出问题的SQL是怎么写的?*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370073?spm=a2c6h.13066369.question.37.33bf585fO2aAwF



问题三:Flink-yarn模块加载外部文件的问题

hi all, 最近在1.11上使用io.extra-file加载外部http文件出现一点小问题

由于http的文件在FileSystem.getFileStatus去拿状态时会走HttpFileSystem的类的getFileStatus方法,该方法返回的FileStatus中length为-1。 在client端校验通过后,在decodeYarnLocalResourceDescriptor的时候会碰到问题。 异常如下: 2020-12-04 17:01:28.955 ERROR org.apache.flink.yarn.YarnResourceManager - Could not start TaskManager in container containerXXXXXX. org.apache.flink.util.FlinkException: Error to parse YarnLocalResourceDescriptor from YarnLocalResourceDescriptor{key=XXXXX.jar, path=https://XXXXXXX.jar, size=-1, modificationTime=0, visibility=APPLICATION} at org.apache.flink.yarn.YarnLocalResourceDescriptor.fromString(YarnLocalResourceDescriptor.java:99) at org.apache.flink.yarn.Utils.decodeYarnLocalResourceDescriptorListFromString(Utils.java:721) at org.apache.flink.yarn.Utils.createTaskExecutorContext(Utils.java:626) at org.apache.flink.yarn.YarnResourceManager.getOrCreateContainerLaunchContext(YarnResourceManager.java:746) at org.apache.flink.yarn.YarnResourceManager.createTaskExecutorLaunchContext(YarnResourceManager.java:726) at org.apache.flink.yarn.YarnResourceManager.startTaskExecutorInContainer(YarnResourceManager.java:500) at org.apache.flink.yarn.YarnResourceManager.onContainersOfResourceAllocated(YarnResourceManager.java:455) at org.apache.flink.yarn.YarnResourceManager.lambda$onContainersAllocated$1(YarnResourceManager.java:415) 。。。。。

因为正则匹配的size只能匹配数字

private static final Pattern LOCAL_RESOURCE_DESC_FORMAT = Pattern.compile("YarnLocalResourceDescriptor\{" + "key=(\S+), path=(\S+), size=([\d]+), modificationTime=([\d]+), visibility=(\S+), type=(\S+)}");

负号不能被匹配上,想要实现加载http的文件需要mock一个length大于-1的FileStatus。 想问下各位大佬,看到有加载remote文件的功能,这块功能是不是给远程hdfs上文件用的,并没有考虑到http文件呢?

*来自志愿者整理的flink邮件归档



参考答案:

你指的是 yarn.provided.lib.dirs 吗? 这个地方设计的时候确实是只考虑到了 hdfs,没有考虑 http 文件。

我刚刚也和熟悉 hadoop 的朋友确认了一下,从 yarn 的层面上应该也是支持 http 文件的,所以 flink 最好也是能支持 http 文件。 由于整个 feature 在设计的时候都没有考虑到 http 文件的问题,不确定修复了正则匹配的问题后是否还会有其他问题。 我开了 FLINK-20505 来跟踪 provided lib 支持 http 文件的整体进度。

https://issues.apache.org/jira/browse/FLINK-20505

*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370074?spm=a2c6h.13066369.question.40.33bf585fcYjaH4



问题四:使用stop命令停止yarn队列任务携带Kerberos认证信息

使用./bin/flink stop -yid application_xxx_xxx xxxxxx命令停止yarn任务时,可否将kerberos认证信息带过去? 例如说,-Dsecurity.kerberos.login.keytab 及 -Dsecurity.kerberos.login.principal这样(这二者试过不行 )

btw, 代码形式的stop和命令行的stop命令使用效率是一致的吗?thanks.*来自志愿者整理的flink邮件归档



参考答案:

先kinit执行把认证注入,再执行停止命令*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370075?spm=a2c6h.13066369.question.39.33bf585f1elY3Q



问题五:Flink1.11.2写hive提示UnknownHostException

环境:Flink1.11.2, CDH5.13.3(Hive1.1.0) 已使用的方法: flink-conf.yaml中设置env.yarn.conf.dir,另外也检测集群配置,hdfs-site.xml没问题,但都无效 异常: Caused by: java.net.UnknownHostException: nameservice at org.apache.hadoop.security.SecurityUtil.buildTokenService(SecurityUtil.java:406) ~[hadoop-common.jar:?] at org.apache.hadoop.hdfs.NameNodeProxies.createNonHAProxy(NameNodeProxies.java:310) ~[hadoop-hdfs.jar:?] at org.apache.hadoop.hdfs.NameNodeProxies.createProxy(NameNodeProxies.java:176) ~[hadoop-hdfs.jar:?] at org.apache.hadoop.hdfs.DFSClient. (DFSClient.java:735) ~[hadoop-hdfs.jar:?] at org.apache.hadoop.hdfs.DFSClient. (DFSClient.java:678) ~[hadoop-hdfs.jar:?] at org.apache.hadoop.hdfs.DistributedFileSystem.initialize(DistributedFileSystem.java:158) ~[hadoop-hdfs.jar:?] at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2816) ~[hadoop-common.jar:?] at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:98) ~[hadoop-common.jar:?] at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2853) ~[hadoop-common.jar:?] at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2835) ~[hadoop-common.jar:?] at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:387) ~[hadoop-common.jar:?] at org.apache.flink.formats.hadoop.bulk.committer.HadoopRenameFileCommitter.generateTempFilePath(HadoopRenameFileCommitter.java:112) ~[flink-connector-hive_2.11-1.11.2.jar:1.11.2] at org.apache.flink.formats.hadoop.bulk.committer.HadoopRenameFileCommitter. (HadoopRenameFileCommitter.java:48) ~[flink-connector-hive_2.11-1.11.2.jar:1.11.2] at org.apache.flink.formats.hadoop.bulk.DefaultHadoopFileCommitterFactory.create(DefaultHadoopFileCommitterFactory.java:40) ~[flink-connector-hive_2.11-1.11.2.jar:1.11.2] at org.apache.flink.formats.hadoop.bulk.HadoopPathBasedPartFileWriter$HadoopPathBasedBucketWriter.openNewInProgressFile(HadoopPathBasedPartFileWriter.java:254) ~[flink-connector-hive_2.11-1.11.2.jar:1.11.2] at org.apache.flink.formats.hadoop.bulk.HadoopPathBasedPartFileWriter$HadoopPathBasedBucketWriter.openNewInProgressFile(HadoopPathBasedPartFileWriter.java:230) ~[flink-connector-hive_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.rollPartFile(Bucket.java:209) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.write(Bucket.java:200) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.onElement(Buckets.java:282) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.onElement(StreamingFileSinkHelper.java:104) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.table.filesystem.stream.StreamingFileWriter.processElement(StreamingFileWriter.java:119) ~[flink-table-blink_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:717) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) ~[flink-dist_2.11-1.11.2.jar:1.11.2]

at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at SourceConversion$3.processElement(Unknown Source) ~[?:?] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:717) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:41) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:717) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:692) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:672) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:52) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.CountingOutput.collect(CountingOutput.java:30) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordsWithTimestamps(AbstractFetcher.java:352) ~[ae3cd858c38340e48cede6e6d9322451.jar:?] at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.partitionConsumerRecordsHandler(KafkaFetcher.java:185) ~[ae3cd858c38340e48cede6e6d9322451.jar:?] at org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:141) ~[ae3cd858c38340e48cede6e6d9322451.jar:?] at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.runWithPartitionDiscovery(FlinkKafkaConsumerBase.java:765) ~[ae3cd858c38340e48cede6e6d9322451.jar:?] at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:757) ~[ae3cd858c38340e48cede6e6d9322451.jar:?] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:101) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63) ~[flink-dist_2.11-1.11.2.jar:1.11.2] at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:213) ~[flink-dist_2.11-1.11.2.jar:1.11.2]

*来自志愿者整理的flink邮件归档



参考答案:

一般UnknownHostException的报错就是因为HDFS client没有解析到正确的hdfs-site.xml配置, 导致把nameservice当成hostname来解析了,你可以看JobManager以及TaskManager的log, 查看里面打出来的classpath有没有hadoop conf的目录*来自志愿者整理的flink邮件归档



关于本问题的更多回答可点击进行查看:

https://developer.aliyun.com/ask/370076?spm=a2c6h.13066369.question.40.33bf585fQDVjlH

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
相关文章
|
23天前
|
Java 关系型数据库 MySQL
Flink1.18.1和CDC2.4.1 本地没问题 提交任务到服务器 报错java.lang.NoClassDefFoundError: Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig
【2月更文挑战第33天】Flink1.18.1和CDC2.4.1 本地没问题 提交任务到服务器 报错java.lang.NoClassDefFoundError: Could not initialize class io.debezium.connector.mysql.MySqlConnectorConfig
35 2
|
27天前
|
Java 关系型数据库 MySQL
Flink CDC有见这个报错不?
【2月更文挑战第29天】Flink CDC有见这个报错不?
17 2
|
30天前
|
存储 关系型数据库 MySQL
Flink CDC产品常见问题之写hudi的时候报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
30天前
|
资源调度 关系型数据库 测试技术
Flink CDC产品常见问题之没有报错但是一直监听不到数据如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
30天前
|
缓存 监控 Java
Flink CDC产品常见问题之flink集群jps命令报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
1月前
|
Oracle 关系型数据库 MySQL
Flink CDC产品常见问题之用superset连接starrocks报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
1月前
|
Oracle 关系型数据库 MySQL
flink cdc 增量问题之增量数据会报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
1月前
|
Oracle 关系型数据库 MySQL
Flink CDC产品常见问题之使用cdc-Oracle连接器报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
Flink CDC产品常见问题之使用cdc-Oracle连接器报错如何解决
|
1月前
|
Oracle 关系型数据库 数据处理
Flink CDC产品常见问题之flink postgresqlcdc 报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。
|
1月前
|
存储 JSON 资源调度
flink cdc 同步问题之出现报错如何解决
Flink CDC(Change Data Capture)是一个基于Apache Flink的实时数据变更捕获库,用于实现数据库的实时同步和变更流的处理;在本汇总中,我们组织了关于Flink CDC产品在实践中用户经常提出的问题及其解答,目的是辅助用户更好地理解和应用这一技术,优化实时数据处理流程。

相关产品

  • 实时计算 Flink版