You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@beam.apache.org by Pawel Bartoszek <pa...@gmail.com> on 2017/09/12 09:10:56 UTC
Unable to find registrar for s3n when restoring flink job from savepoint
Hi,
I am running a flink v1.2.1 job on a EMR cluster using Beam v2.0.0.
When I try to restore job from savepoint on one task manager I get the
exception *Unable to find registrar for s3n*.
The job can write files to s3 acting as a sink. So S3 access works except
when restoring from savepoint.
I am passing the following configuration as the pipeline
HadoopFileSystemOptions
options:
Configuration configuration = new Configuration();
configuration.set("fs.defaultFS", String.format("s3n://%s",
jobOptions.getOutputFileSystemRoot()));
configuration.set("fs.default.name", String.format("s3n://%s",
jobOptions.getOutputFileSystemRoot()));
configuration.set("fs.s3.impl",
"org.apache.hadoop.fs.s3native.NativeS3FileSystem");
configuration.set("fs.s3n.awsAccessKeyId",
jobOptions.getAwsAccessKey());
configuration.set("fs.s3n.awsSecretAccessKey",
jobOptions.getAwsSecretKey());
From my investigation it looks like Beam FileSystems.
setDefaultPipelineOptions method is not called before
org.apache.beam.sdk.io.FileBasedSink$FileResultCoder relay on initialised
FileSystems.SCHEME_TO_FILESYSTEM map
java.lang.IllegalStateException: Could not initialize keyed state backend.
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initKeyedState(AbstractStreamOperator.java:293)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:204)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeOperators(StreamTask.java:653)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:640)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:246)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:665)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalStateException: Unable to find registrar for s3n
at
org.apache.beam.sdk.io.FileSystems.getFileSystemInternal(FileSystems.java:447)
at org.apache.beam.sdk.io.FileSystems.matchNewResource(FileSystems.java:523)
at
org.apache.beam.sdk.io.FileBasedSink$FileResultCoder.decode(FileBasedSink.java:1059)
at
org.apache.beam.sdk.io.FileBasedSink$FileResultCoder.decode(FileBasedSink.java:1020)
at
org.apache.beam.runners.flink.translation.types.CoderTypeSerializer.deserialize(CoderTypeSerializer.java:87)
at
org.apache.flink.runtime.state.ArrayListSerializer.deserialize(ArrayListSerializer.java:87)
at
org.apache.flink.runtime.state.ArrayListSerializer.deserialize(ArrayListSerializer.java:27)
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.readStateTableForKeyGroup(HeapKeyedStateBackend.java:370)
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restorePartitionedState(HeapKeyedStateBackend.java:340)
at
org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(HeapKeyedStateBackend.java:243)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.createKeyedStateBackend(StreamTask.java:788)
at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initKeyedState(AbstractStreamOperator.java:284)
... 6 more
Pawel Bartoszek
Re: Unable to find registrar for s3n when restoring flink job from
savepoint
Posted by Aljoscha Krettek <al...@apache.org>.
I responded on the issue.
> On 12. Sep 2017, at 22:25, Lukasz Cwik <lc...@google.com.INVALID> wrote:
>
> Filed https://issues.apache.org/jira/browse/BEAM-2948
>
> On Tue, Sep 12, 2017 at 2:10 AM, Pawel Bartoszek <pawelbartoszek89@gmail.com
>> wrote:
>
>> Hi,
>>
>> I am running a flink v1.2.1 job on a EMR cluster using Beam v2.0.0.
>> When I try to restore job from savepoint on one task manager I get the
>> exception *Unable to find registrar for s3n*.
>> The job can write files to s3 acting as a sink. So S3 access works except
>> when restoring from savepoint.
>>
>> I am passing the following configuration as the pipeline
>> HadoopFileSystemOptions
>> options:
>>
>> Configuration configuration = new Configuration();
>> configuration.set("fs.defaultFS", String.format("s3n://%s",
>> jobOptions.getOutputFileSystemRoot()));
>> configuration.set("fs.default.name", String.format("s3n://%s",
>> jobOptions.getOutputFileSystemRoot()));
>> configuration.set("fs.s3.impl",
>> "org.apache.hadoop.fs.s3native.NativeS3FileSystem");
>> configuration.set("fs.s3n.awsAccessKeyId",
>> jobOptions.getAwsAccessKey());
>> configuration.set("fs.s3n.awsSecretAccessKey",
>> jobOptions.getAwsSecretKey());
>>
>>
>>
>> From my investigation it looks like Beam FileSystems.
>> setDefaultPipelineOptions method is not called before
>> org.apache.beam.sdk.io.FileBasedSink$FileResultCoder relay on initialised
>> FileSystems.SCHEME_TO_FILESYSTEM map
>>
>>
>>
>> java.lang.IllegalStateException: Could not initialize keyed state backend.
>> at
>> org.apache.flink.streaming.api.operators.AbstractStreamOperator.
>> initKeyedState(AbstractStreamOperator.java:293)
>> at
>> org.apache.flink.streaming.api.operators.AbstractStreamOperator.
>> initializeState(AbstractStreamOperator.java:204)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.initializeOperators(
>> StreamTask.java:653)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.
>> initializeState(StreamTask.java:640)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.
>> invoke(StreamTask.java:246)
>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:665)
>> at java.lang.Thread.run(Thread.java:748)
>> Caused by: java.lang.IllegalStateException: Unable to find registrar for
>> s3n
>> at
>> org.apache.beam.sdk.io.FileSystems.getFileSystemInternal(
>> FileSystems.java:447)
>> at org.apache.beam.sdk.io.FileSystems.matchNewResource(
>> FileSystems.java:523)
>> at
>> org.apache.beam.sdk.io.FileBasedSink$FileResultCoder.
>> decode(FileBasedSink.java:1059)
>> at
>> org.apache.beam.sdk.io.FileBasedSink$FileResultCoder.
>> decode(FileBasedSink.java:1020)
>> at
>> org.apache.beam.runners.flink.translation.types.CoderTypeSerializer.
>> deserialize(CoderTypeSerializer.java:87)
>> at
>> org.apache.flink.runtime.state.ArrayListSerializer.deserialize(
>> ArrayListSerializer.java:87)
>> at
>> org.apache.flink.runtime.state.ArrayListSerializer.deserialize(
>> ArrayListSerializer.java:27)
>> at
>> org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.
>> readStateTableForKeyGroup(HeapKeyedStateBackend.java:370)
>> at
>> org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.
>> restorePartitionedState(HeapKeyedStateBackend.java:340)
>> at
>> org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(
>> HeapKeyedStateBackend.java:243)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.
>> createKeyedStateBackend(StreamTask.java:788)
>> at
>> org.apache.flink.streaming.api.operators.AbstractStreamOperator.
>> initKeyedState(AbstractStreamOperator.java:284)
>> ... 6 more
>>
>>
>>
>> Pawel Bartoszek
>>
Re: Unable to find registrar for s3n when restoring flink job from savepoint
Posted by Lukasz Cwik <lc...@google.com.INVALID>.
Filed https://issues.apache.org/jira/browse/BEAM-2948
On Tue, Sep 12, 2017 at 2:10 AM, Pawel Bartoszek <pawelbartoszek89@gmail.com
> wrote:
> Hi,
>
> I am running a flink v1.2.1 job on a EMR cluster using Beam v2.0.0.
> When I try to restore job from savepoint on one task manager I get the
> exception *Unable to find registrar for s3n*.
> The job can write files to s3 acting as a sink. So S3 access works except
> when restoring from savepoint.
>
> I am passing the following configuration as the pipeline
> HadoopFileSystemOptions
> options:
>
> Configuration configuration = new Configuration();
> configuration.set("fs.defaultFS", String.format("s3n://%s",
> jobOptions.getOutputFileSystemRoot()));
> configuration.set("fs.default.name", String.format("s3n://%s",
> jobOptions.getOutputFileSystemRoot()));
> configuration.set("fs.s3.impl",
> "org.apache.hadoop.fs.s3native.NativeS3FileSystem");
> configuration.set("fs.s3n.awsAccessKeyId",
> jobOptions.getAwsAccessKey());
> configuration.set("fs.s3n.awsSecretAccessKey",
> jobOptions.getAwsSecretKey());
>
>
>
> From my investigation it looks like Beam FileSystems.
> setDefaultPipelineOptions method is not called before
> org.apache.beam.sdk.io.FileBasedSink$FileResultCoder relay on initialised
> FileSystems.SCHEME_TO_FILESYSTEM map
>
>
>
> java.lang.IllegalStateException: Could not initialize keyed state backend.
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.
> initKeyedState(AbstractStreamOperator.java:293)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.
> initializeState(AbstractStreamOperator.java:204)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.initializeOperators(
> StreamTask.java:653)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.
> initializeState(StreamTask.java:640)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.
> invoke(StreamTask.java:246)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:665)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: java.lang.IllegalStateException: Unable to find registrar for
> s3n
> at
> org.apache.beam.sdk.io.FileSystems.getFileSystemInternal(
> FileSystems.java:447)
> at org.apache.beam.sdk.io.FileSystems.matchNewResource(
> FileSystems.java:523)
> at
> org.apache.beam.sdk.io.FileBasedSink$FileResultCoder.
> decode(FileBasedSink.java:1059)
> at
> org.apache.beam.sdk.io.FileBasedSink$FileResultCoder.
> decode(FileBasedSink.java:1020)
> at
> org.apache.beam.runners.flink.translation.types.CoderTypeSerializer.
> deserialize(CoderTypeSerializer.java:87)
> at
> org.apache.flink.runtime.state.ArrayListSerializer.deserialize(
> ArrayListSerializer.java:87)
> at
> org.apache.flink.runtime.state.ArrayListSerializer.deserialize(
> ArrayListSerializer.java:27)
> at
> org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.
> readStateTableForKeyGroup(HeapKeyedStateBackend.java:370)
> at
> org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.
> restorePartitionedState(HeapKeyedStateBackend.java:340)
> at
> org.apache.flink.runtime.state.heap.HeapKeyedStateBackend.restore(
> HeapKeyedStateBackend.java:243)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.
> createKeyedStateBackend(StreamTask.java:788)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.
> initKeyedState(AbstractStreamOperator.java:284)
> ... 6 more
>
>
>
> Pawel Bartoszek
>