需要将flink-hadoop-compatibility-2.10 jar显式复制到EMR集群上的${FLINK-HOME}/lib位置

Question

我目前正在开发一个Flink应用程序，它使用一些Hadoop依赖项将数据写入S3位置。在本地环境中，它工作得很好，但是当我在EMR集群上部署这个Flink应用程序时，它引发了一个与兼容性问题有关的异常。

我得到的错误消息是

java.lang.RuntimeException：无法加载类'org.apache.hadoop.io.writable'的TypeInformation。您可能缺少“flink-hadoop-compatibility”依赖项。在org.apache.flink.api.java.typeutils.typeExtractor.createHadoopWritableTypeInfo（typeExtractor.java:2025)在org.apache.flink.api.java.typeExtractor.typeExtractor.privateGetForClass（typeExtractor.java:1649)在org.apache.flink.api.java.typeExtractor.typeExtractor.java.privateGetForClass（

我在POM依赖项中包含了flink-hadoop-compatibility-2.10jar的maven依赖项。但它没有探测到它。我使用的Flink版本是1.2.0

但是，当我显式地将兼容性JAR复制到${FLINK-HOME}/lib位置时，我不会得到任何异常，并且能够成功地运行Flink应用程序。

我们是否可以使用任何方法，以便在不将JAR文件部署到${FLINK-HOME}/lib的情况下运行应用程序？

在POM依赖项中需要什么修改，以便应用程序能够检测到它，并且不需要将兼容性JAR复制到flink-home/lib位置？

Answer 1

package org.apache.flink.api.java.typeutils;

public class TypeExtractor {

    /** The name of the class representing Hadoop's writable */
    private static final String HADOOP_WRITABLE_CLASS = "org.apache.hadoop.io.Writable";
    private static final String HADOOP_WRITABLE_TYPEINFO_CLASS = "org.apache.flink.api.java.typeutils.WritableTypeInfo";


    // visible for testing
    public static <T> TypeInformation<T> createHadoopWritableTypeInfo(Class<T> clazz) {
        checkNotNull(clazz);

        Class<?> typeInfoClass;
        try {
            typeInfoClass = Class.forName(HADOOP_WRITABLE_TYPEINFO_CLASS, false, TypeExtractor.class.getClassLoader());
        }
        catch (ClassNotFoundException e) {
            throw new RuntimeException("Could not load the TypeInformation for the class '"
                    + HADOOP_WRITABLE_CLASS + "'. You may be missing the 'flink-hadoop-compatibility' dependency.");
        }

        ...
    }
}

这是因为org.apache.hadoop.io.writable意味着由typeextractor.class.getClassLoader（）加载，它是appClassLoader，而子项flink jar是由parentFirstClassLoader加载的，它是appClassLoader的子项，因此appClassLoader不能从flink jar加载

我不确定这是否是一个bug，将classLoader改为thread.currentThread().getContextClassLoader()将使其工作，而无需将flink-hadoop-compatibility jar文件复制到${FLINK-HOME}/lib位置。