问题一：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
