Uruk & Igor, Thanks for helping out! Yup, it fixed my issue.
Chen On Wed, May 4, 2016 at 12:57 PM, Igor Berman <igor.ber...@gmail.com> wrote: > I think I've had this issue too and fixed it as Ufuk suggested > in core-site.xml > > something like > <property> > <name>fs.s3a.buffer.dir</name> > <value>/tmp</value> > </property> > > > On 4 May 2016 at 11:10, Ufuk Celebi <u...@apache.org> wrote: > >> Hey Chen Qin, >> >> this seems to be an issue with the S3 file system. The root cause is: >> >> Caused by: java.lang.NullPointerException at >> >> org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.confChanged(LocalDirAllocator.java:268) >> at >> org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.getLocalPathForWrite(LocalDirAllocator.java:344) >> at >> org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.createTmpFileForWrite(LocalDirAllocator.java:416) >> at >> org.apache.hadoop.fs.LocalDirAllocator.createTmpFileForWrite(LocalDirAllocator.java:198) >> at >> org.apache.hadoop.fs.s3a.S3AOutputStream.<init>(S3AOutputStream.java:87) >> at org.apache.hadoop.fs.s3a.S3AFileSystem.create(S3AFileSystem.java:410) >> at org.apache.hadoop.fs.FileSystem.create(FileSystem.java:907) at >> org.apache.hadoop.fs.FileSystem.create(FileSystem.java:888) at >> org.apache.hadoop.fs.FileSystem.create(FileSystem.java:785) at >> >> org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.create(HadoopFileSystem.java:404) >> at >> org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.create(HadoopFileSystem.java:48) >> at >> org.apache.flink.runtime.state.filesystem.FsStateBackend$FsCheckpointStateOutputStream.flush(FsStateBackend.java:489) >> ... 25 more >> >> From [1] it looks like you have to specify >> >> fs.s3a.buffer.dir >> >> in the Hadoop configuration (where you set the S3 file system). >> >> The expected value is a comma separated list of local directories used >> to buffer results prior to transmitting the to S3 (for large files). >> >> Does this fix the issue? Please report back so that we can include in >> the "common issues" section of the AWS docs. >> >> – Ufuk >> >> [1] http://deploymentzone.com/2015/12/20/s3a-on-spark-on-aws-ec2/ >> >> >> On Wed, May 4, 2016 at 2:41 AM, Chen Qin <qinnc...@gmail.com> wrote: >> > Hi there, >> > >> > I run a test job with filestatebackend and save checkpoints on s3 (via >> s3a) >> > >> > The job crash when checkpoint triggered. Looking into s3 directory and >> list >> > objects. I found the directory is create successfully but all >> checkpoints >> > directory size are empty. >> > >> > The host running task manager shows following error. >> > >> > Received error response: >> com.amazonaws.services.s3.model.AmazonS3Exception: >> > Status Code: 404, AWS Service: null, AWS Request ID: CF1845CA84E07549, >> AWS >> > Error Code: null, AWS Error Message: Not Found, S3 Extended Request >> ID:xxxxx >> > >> > Has anyone met this issue before? >> > >> > flink 1.0.0 >> > scala 2.10 >> > hadoop-aws 2.7.2 >> > aws-java-sdk 1.7.4 >> > >> > >> > Thanks, >> > Chen >> > >> > Attached full log that shows on web dashboard when job canceled. >> > java.lang.RuntimeException: Error triggering a checkpoint as the result >> of >> > receiving checkpoint barrier at >> > >> org.apache.flink.streaming.runtime.tasks.StreamTask$2.onEvent(StreamTask.java:681) >> > at >> > >> org.apache.flink.streaming.runtime.tasks.StreamTask$2.onEvent(StreamTask.java:674) >> > at >> > >> org.apache.flink.streaming.runtime.io.BarrierBuffer.processBarrier(BarrierBuffer.java:203) >> > at >> > >> org.apache.flink.streaming.runtime.io.BarrierBuffer.getNextNonBlocked(BarrierBuffer.java:129) >> > at >> > >> org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:175) >> > at >> > >> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:65) >> > at >> > >> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:224) >> > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559) at >> > java.lang.Thread.run(Thread.java:745) Caused by: java.io.IOException: >> Could >> > not open output stream for state backend at >> > >> org.apache.flink.runtime.state.filesystem.FsStateBackend$FsCheckpointStateOutputStream.flush(FsStateBackend.java:498) >> > at >> > >> org.apache.flink.runtime.state.filesystem.FsStateBackend$FsCheckpointStateOutputStream.write(FsStateBackend.java:444) >> > at java.io.DataOutputStream.write(DataOutputStream.java:88) at >> > java.io.DataOutputStream.write(DataOutputStream.java:88) at >> > org.apache.flink.types.StringValue.writeString(StringValue.java:813) at >> > >> org.apache.flink.api.common.typeutils.base.StringSerializer.serialize(StringSerializer.java:64) >> > at >> > >> org.apache.flink.api.common.typeutils.base.StringSerializer.serialize(StringSerializer.java:28) >> > at >> > >> org.apache.flink.api.java.typeutils.runtime.TupleSerializer.serialize(TupleSerializer.java:124) >> > at >> > >> org.apache.flink.api.java.typeutils.runtime.TupleSerializer.serialize(TupleSerializer.java:30) >> > at >> > >> org.apache.flink.runtime.state.ArrayListSerializer.serialize(ArrayListSerializer.java:78) >> > at >> > >> org.apache.flink.runtime.state.ArrayListSerializer.serialize(ArrayListSerializer.java:27) >> > at >> > >> org.apache.flink.runtime.state.filesystem.AbstractFsState.snapshot(AbstractFsState.java:85) >> > at >> > >> org.apache.flink.runtime.state.AbstractStateBackend.snapshotPartitionedState(AbstractStateBackend.java:265) >> > at >> > >> org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotOperatorState(AbstractStreamOperator.java:175) >> > at >> > >> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.snapshotOperatorState(AbstractUdfStreamOperator.java:121) >> > at >> > >> org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.snapshotOperatorState(WindowOperator.java:509) >> > at >> > >> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoint(StreamTask.java:481) >> > at >> > >> org.apache.flink.streaming.runtime.tasks.StreamTask$2.onEvent(StreamTask.java:678) >> > ... 8 more Caused by: java.lang.NullPointerException at >> > >> org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.confChanged(LocalDirAllocator.java:268) >> > at >> > >> org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.getLocalPathForWrite(LocalDirAllocator.java:344) >> > at >> > >> org.apache.hadoop.fs.LocalDirAllocator$AllocatorPerContext.createTmpFileForWrite(LocalDirAllocator.java:416) >> > at >> > >> org.apache.hadoop.fs.LocalDirAllocator.createTmpFileForWrite(LocalDirAllocator.java:198) >> > at >> org.apache.hadoop.fs.s3a.S3AOutputStream.<init>(S3AOutputStream.java:87) >> > at >> org.apache.hadoop.fs.s3a.S3AFileSystem.create(S3AFileSystem.java:410) at >> > org.apache.hadoop.fs.FileSystem.create(FileSystem.java:907) at >> > org.apache.hadoop.fs.FileSystem.create(FileSystem.java:888) at >> > org.apache.hadoop.fs.FileSystem.create(FileSystem.java:785) at >> > >> org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.create(HadoopFileSystem.java:404) >> > at >> > >> org.apache.flink.runtime.fs.hdfs.HadoopFileSystem.create(HadoopFileSystem.java:48) >> > at >> > >> org.apache.flink.runtime.state.filesystem.FsStateBackend$FsCheckpointStateOutputStream.flush(FsStateBackend.java:489) >> > ... 25 more >> > >> > >
