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

Question

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

我得到的错误信息是

java.lang.RuntimeException: Could not load the TypeInformation for the class 'org.apache.hadoop.io.Writable'. You may be missing the 'flink-hadoop-compatibility' dependency. at org.apache.flink.api.java.typeutils.TypeExtractor.createHadoopWritableTypeInfo(TypeExtractor.java:2025) at org.apache.flink.api.java.typeutils.TypeExtractor.privateGetForClass(TypeExtractor.java:1649) at org.apache.flink.api.java.typeutils.TypeExtractor.privateGetForClass(TypeExtractor.java:1591) at org.apache.flink.api.java.typeutils.TypeExtractor.createTypeInfoWithTypeHierarchy(TypeExtractor.java:778) ....

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

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

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

或

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

Answer 1

在查看了各种帖子并对POM文件进行了实验之后，我认为在当前版本的Apache Flink (1.2.0)中，需要将JAR文件复制(部署)到${FLINK-HOME}/lib位置。

Answer 2

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 )加载，而子链接jar是由ParentFirstClassLoader加载的，ParentFirstClassLoader是AppClassLoader的子级，因此AppClassLoader无法从flink jar加载org.apache.hadoop.io.Writable。

我不确定这是否是一个bug，如果不将FLINK-hadoop兼容jar文件复制到${FLINK-HOME} classLoader位置，则切换到Thread.currentThread().getContextClassLoader()将使其工作。