为Apache-Flink提交作业时的Amazon EMR在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的org.apache.flink.runtime.fs.hdfs.HadoopRecoverableWriter。（HadoopRecoverableWriter.java:57）。 .runtime.fs.hdfs.HadoopFileSystem.createRecoverableWriter（HadoopFileSystem.java:202）位于org.apache.flink.flink.stream.stream./循环函数.sink.filesystem.Buckets。（Buckets.java:112）位于org.apache.flink的org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink $ RowFormatBuilder.createBuckets（StreamingFileSink.java:242）。 streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState（StreamingFileSink.java:327）位于org.apache.flink的org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction（StreamingFunctionUtils.java:178）。 streaming.util.functions.StreamingFunctionUtils.restoreFunctionSta te（StreamingFunctionUtils.java:160）org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState（AbstractUdfStreamOperator.java:96）at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState（AbstractStreamOperator。 java：278）org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState（StreamTask.java:738）at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke（StreamTask.java:289） at org.apache.flink.runtime.taskmanager.Task.run（Task.java:704）at java.lang.Thread.run（Thread.java:748）

Answer 1

为了使用Flink的StreamingFileSink，只需要一次保证，你需要使用Hadoop> = 2.7。不支持2.7以下的版本。因此，请确保您在EMR上运行最新的Hadoop版本。