我有一个简单的flink作业(flink版本1.8.3),它有一个每10分钟写入s3存储桶的接收器。
当我停止工作,并重新启动它从以前的快照我经常得到这个错误,阻止我继续正确的工作
"throwableInformation": [
"java.io.FileNotFoundException: No such file or directory: s3://my-bucket/2020-09-22--18/_part-0-0_tmp_91f865fe-4774-4699-80ca-483aa36e3fd9",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2231)",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2149)",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2088)",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.open(S3AFileSystem.java:699)",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.FileSystem.open(FileSystem.java:950)",
"\tat org.apache.flink.fs.s3hadoop.HadoopS3AccessHelper.getObject(HadoopS3AccessHelper.java:98)",
"\tat org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverInProgressPart(S3RecoverableMultipartUploadFactory.java:97)",
"\tat org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverRecoverableUpload(S3RecoverableMultipartUploadFactory.java:75)",
"\tat org.apache.flink.fs.s3.common.writer.S3RecoverableWriter.recover(S3RecoverableWriter.java:95)",
"\tat org.apache.flink.fs.s3.common.writer.S3RecoverableWriter.recover(S3RecoverableWriter.java:50)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.restoreInProgressFile(Bucket.java:140)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.<init>(Bucket.java:127)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.restore(Bucket.java:399)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.DefaultBucketFactoryImpl.restoreBucket(DefaultBucketFactoryImpl.java:64)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.handleRestoredBucketState(Buckets.java:177)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.initializeActiveBuckets(Buckets.java:165)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.initializeState(Buckets.java:149)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState(StreamingFileSink.java:334)",
"\tat org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:178)",
"\tat org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:160)",
"\tat org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96)",
"\tat org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:278)",
"\tat org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:763)",
"\tat org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:297)",
"\tat org.apache.flink.runtime.taskmanager.Task.run(Task.java:714)",
"\tat java.lang.Thread.run(Thread.java:748)"
],
......
......) -> to: Result -> (Sink: Print to Std. Out, Sink: Unnamed, Sink: Unnamed, Sink: Unnamed) (1/1) (7dc2e0b7013bbfddf507531ebc184c46) switched from RUNNING to FAILED.","throwableInformation":["java.io.FileNotFoundException: No such file or directory: s3://my-bucket/2020-09-22--18/_part-0-0_tmp_91f865fe-4774-4699-80ca-483aa36e3fd9","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2231)","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2149)","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2088)","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.open(S3AFileSystem.java:699)","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.FileSystem.open(FileSystem.java:950)","\tat org.apache.flink.fs.s3hadoop.HadoopS3AccessHelper.getObject(HadoopS3AccessHelper.java:98)","\tat org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverInProgressPart(S3RecoverableMultipartUploadFactory.java:97)","\tat org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverRecoverableUpload(S3RecoverableMultipartUploadFactory.java:75)","\tat org.apache.flink.fs.s3.common.writer.S3RecoverableWriter.recover(S3RecoverableWriter.java:95)","\tat org.apache.flink.fs.s3.common.writer.S3RecoverableWriter.recover(S3RecoverableWriter.java:50)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.restoreInProgressFile(Bucket.java:140)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.<init>(Bucket.java:127)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.restore(Bucket.java:399)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.DefaultBucketFactoryImpl.restoreBucket(DefaultBucketFactoryImpl.java:64)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.handleRestoredBucketState(Buckets.java:177)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.initializeActiveBuckets(Buckets.java:165)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.initializeState(Buckets.java:149)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState(StreamingFileSink.java:334)","\tat org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:178)","\tat org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:160)","\tat org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96)","\tat org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:278)","\tat org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:763)","\tat
我已经签入了s3 bucket,但是临时文件不在那里。它看起来像是,当作业停止时,在写入文件时出现了某种错误,当我重新启动作业时,它会再次尝试查找该文件,但它不在那里?
我怎样才能避免这个问题?有没有办法忽略丢失的文件?
暂无答案!
目前还没有任何答案,快来回答吧!