Amazon EMR при отправке задания на Apache-Flink получает ошибку с исправлением Hadoop

Question

Added Depedency Pom Details :

<dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>1.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>1.7.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>1.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime_2.11</artifactId>
            <version>1.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table_2.11</artifactId>
            <version>1.7.1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
            <version>1.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-filesystem_2.11</artifactId>
            <version>1.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-hadoop-compatibility_2.11</artifactId>
            <version>1.7.1</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-s3-fs-hadoop</artifactId>
            <version>1.7.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-shaded-hadoop</artifactId>
            <version>1.7.1</version>
            <type>pom</type>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-aws</artifactId>
            <version>2.8.5</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-hdfs</artifactId>
            <version>2.8.5</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>2.8.5</version>
        </dependency>
        <dependency>
            <groupId>com.amazonaws</groupId>
            <artifactId>aws-java-sdk-s3</artifactId>
            <version>1.11.529</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-connectors</artifactId>
            <version>1.1.5</version>
            <type>pom</type>
        </dependency>
    </dependencies>

java.lang.UnsupportedOperationException: восстанавливаемые авторы на Hadoop поддерживаются только для HDFS и Hadoop версии 2.7 или новее в org.apache.flink.runtime.fs.hdfs.HadoopRecoverableWriter. (HadoopRecoverableWriter.java:57) в org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.createRecoverableWriter (HadoopFileSystem.java:202) в org.apache.flink.core.fs.SafetyNetWrapperFileSystem.createRecoverableWriter (SafetyNetWrapperFileSystem.java:69) в org.apache.flink.streaming.api.functions.sink.filesystem.Buckets. (Buckets.java:112) в org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink $ RowFormatBuilder.createBuckets (StreamingFileSink.java:242) в org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState (StreamingFileSink.java:327) в org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction (StreamingFunctionUtils.java:178) в org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState (StreamingFunctionUtils.java:160) в org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState (AbstractUdfStreamOperator.java:96) в org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState (AbstractStreamOperator.java:278) в org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState (StreamTask.java:738) в org.apache.flink.streaming.runtime.tasks.StreamTask.invoke (StreamTask.java:289) в org.apache.flink.runtime.taskmanager.Task.run (Task.java:704) в java.lang.Thread.run (Thread.java:748)

Answer 1

Чтобы использовать Flink's StreamingFileSink с единовременными гарантиями, вам нужно использовать Hadoop> = 2.7. Версии ниже 2.7 не поддерживаются. Поэтому, пожалуйста, убедитесь, что вы используете последнюю версию Hadoop на EMR.