You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Chen Qin <qi...@gmail.com> on 2016/05/16 21:52:55 UTC
rocksdb backend on s3 window operator checkpoint issue
Hi there,
I have been testing checkpointing on rocksdb backed by s3. Checkpoints
seems successful except snapshot states of timeWindow operator on
keyedStream. Here is the env setting I used
env.setStateBackend(new RocksDBStateBackend(new URI("s3://backupdir/")))
The checkpoint for always fail consistently when it goes to window operator
snapshotting. Exception log attached below.
I tried to env.setStateBackend(new RocksDBStateBackend(new URI(
"file:///tmp/checkpoints"))); or MemoryStateBackend(default) works no issue
with checkpoints.
Does anyone saw this issue before? Or did I mess up with configuration?
Thanks,
Chen
2016-05-16 17:20:32,132 INFO
org.apache.flink.runtime.state.filesystem.FsStateBackend -
Initializing file state backend to URI
s3://xxx/checkpoints/7e6abf126ce3d18f173733b34eda81a9
2016-05-16 17:20:32,423 INFO
org.apache.flink.streaming.runtime.tasks.StreamTask - Using
user-defined state backend:
org.apache.flink.contrib.streaming.state.RocksDBStateBackend@2fa68a53
2016-05-16 17:20:32,423 INFO
org.apache.flink.runtime.state.filesystem.FsStateBackend -
Initializing file state backend to URI
s3://uber-beats/sjc1/checkpoints/7e6abf126ce3d18f173733b34eda81a9
2016-05-16 17:21:31,423 INFO
org.apache.flink.contrib.streaming.state.AbstractRocksDBState - RocksDB
(/directory/flink-io-723a5c14-2a8a-4abc-881a-9a60138816b0/7e6abf126ce3d18f173733b34eda81a9/WindowOperator_131_0/window-contents/dbc64864de-8373-4b41-bd74-a26a8007f066)
backup (synchronous part) took 8 ms.
2016-05-16 17:21:36,125 ERROR
org.apache.flink.streaming.runtime.tasks.StreamTask - Caught
exception while materializing asynchronous checkpoints.
com.amazonaws.AmazonClientException: Unable to calculate MD5
hash:/directory//flink-io-723a5c14-2a8a-4abc-881a-9a60138816b0/7e6abf126ce3d18f173733b34eda81a9/WindowOperator_131_0/window-contents/local-chk-599
(Is a directory)
at
com.amazonaws.services.s3.AmazonS3Client.putObject(AmazonS3Client.java:1298)
at
com.amazonaws.services.s3.transfer.internal.UploadCallable.uploadInOneChunk(UploadCallable.java:108)
at
com.amazonaws.services.s3.transfer.internal.UploadCallable.call(UploadCallable.java:100)
at
com.amazonaws.services.s3.transfer.internal.UploadMonitor.upload(UploadMonitor.java:192)
at
com.amazonaws.services.s3.transfer.internal.UploadMonitor.call(UploadMonitor.java:150)
at
com.amazonaws.services.s3.transfer.internal.UploadMonitor.call(UploadMonitor.java:50)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
...
Tests look like
.setParallelism(1).assignTimestampsAndWatermarks(new
AssignerWithPunctuatedWatermarks<String>() {
@Override
public Watermark checkAndGetNextWatermark(String s, long l) {
long ts = System.currentTimeMillis() - 60*1000l;
return new Watermark(ts);
}
@Override
public long extractTimestamp(String s, long l) {
long ts = System.currentTimeMillis();
return ts;
}
}).flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
@Override
public void flatMap(String s, Collector<Tuple2<String, Long>>
collector) throws Exception {
collector.collect(new Tuple2<>(s, 1l));
}
}).keyBy(0).timeWindow(Time.seconds(60)).apply(new
RichWindowFunction<Tuple2<String,Long>, Tuple2<String, Long>, Tuple,
TimeWindow>() {
@Override
public void apply(Tuple tuple, TimeWindow timeWindow,
Iterable<Tuple2<String, Long>> iterable, Collector<Tuple2<String,
Long>> collector) throws Exception {
log.info("trigger fire at ", System.currentTimeMillis());
collector.collect(new
Tuple2<>(String.valueOf(timeWindow.toString()), 1l));
}
}).rebalance().addSink(new FakeSink<>());
JobExecutionResult result = env.execute();
Re: rocksdb backend on s3 window operator checkpoint issue
Posted by Robert Metzger <rm...@apache.org>.
I tried reproducing the issue using the
org.apache.hadoop.fs.s3a.S3AFileSystem and it worked.
I had some dependency issues with the S3AFileSystem so I didn't follow that
path for now. If you've used the S3AFileSystem, I can try to get that one
working as well.
On Tue, May 17, 2016 at 11:59 AM, Robert Metzger <rm...@apache.org>
wrote:
> Hi,
>
> from the code you've provided, everything seems to look okay. I'm
> currently trying to reproduce the issue.
> Which Flink version are you using?
>
> Which s3 implementation did you configure in the hadoop configuration?
>
> Regards,
> Robert
>
>
> On Mon, May 16, 2016 at 11:52 PM, Chen Qin <qi...@gmail.com> wrote:
>
>> Hi there,
>>
>> I have been testing checkpointing on rocksdb backed by s3. Checkpoints
>> seems successful except snapshot states of timeWindow operator on
>> keyedStream. Here is the env setting I used
>> env.setStateBackend(new RocksDBStateBackend(new URI("s3://backupdir/")))
>>
>> The checkpoint for always fail consistently when it goes to window
>> operator snapshotting. Exception log attached below.
>> I tried to env.setStateBackend(new RocksDBStateBackend(new URI(
>> "file:///tmp/checkpoints"))); or MemoryStateBackend(default) works no
>> issue with checkpoints.
>>
>> Does anyone saw this issue before? Or did I mess up with configuration?
>>
>> Thanks,
>> Chen
>>
>> 2016-05-16 17:20:32,132 INFO
>> org.apache.flink.runtime.state.filesystem.FsStateBackend -
>> Initializing file state backend to URI
>> s3://xxx/checkpoints/7e6abf126ce3d18f173733b34eda81a9
>> 2016-05-16 17:20:32,423 INFO
>> org.apache.flink.streaming.runtime.tasks.StreamTask - Using
>> user-defined state backend:
>> org.apache.flink.contrib.streaming.state.RocksDBStateBackend@2fa68a53
>> 2016-05-16 17:20:32,423 INFO
>> org.apache.flink.runtime.state.filesystem.FsStateBackend -
>> Initializing file state backend to URI
>> s3://uber-beats/sjc1/checkpoints/7e6abf126ce3d18f173733b34eda81a9
>> 2016-05-16 17:21:31,423 INFO
>> org.apache.flink.contrib.streaming.state.AbstractRocksDBState - RocksDB
>> (/directory/flink-io-723a5c14-2a8a-4abc-881a-9a60138816b0/7e6abf126ce3d18f173733b34eda81a9/WindowOperator_131_0/window-contents/dbc64864de-8373-4b41-bd74-a26a8007f066)
>> backup (synchronous part) took 8 ms.
>> 2016-05-16 17:21:36,125 ERROR
>> org.apache.flink.streaming.runtime.tasks.StreamTask - Caught
>> exception while materializing asynchronous checkpoints.
>> com.amazonaws.AmazonClientException: Unable to calculate MD5
>> hash:/directory//flink-io-723a5c14-2a8a-4abc-881a-9a60138816b0/7e6abf126ce3d18f173733b34eda81a9/WindowOperator_131_0/window-contents/local-chk-599
>> (Is a directory)
>> at
>> com.amazonaws.services.s3.AmazonS3Client.putObject(AmazonS3Client.java:1298)
>> at
>> com.amazonaws.services.s3.transfer.internal.UploadCallable.uploadInOneChunk(UploadCallable.java:108)
>> at
>> com.amazonaws.services.s3.transfer.internal.UploadCallable.call(UploadCallable.java:100)
>> at
>> com.amazonaws.services.s3.transfer.internal.UploadMonitor.upload(UploadMonitor.java:192)
>> at
>> com.amazonaws.services.s3.transfer.internal.UploadMonitor.call(UploadMonitor.java:150)
>> at
>> com.amazonaws.services.s3.transfer.internal.UploadMonitor.call(UploadMonitor.java:50)
>> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>> at
>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>>
>> ...
>>
>>
>> Tests look like
>>
>> .setParallelism(1).assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<String>() {
>> @Override
>> public Watermark checkAndGetNextWatermark(String s, long l) {
>> long ts = System.currentTimeMillis() - 60*1000l;
>> return new Watermark(ts);
>> }
>>
>> @Override
>> public long extractTimestamp(String s, long l) {
>> long ts = System.currentTimeMillis();
>> return ts;
>> }
>> }).flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
>> @Override
>> public void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {
>> collector.collect(new Tuple2<>(s, 1l));
>> }
>> }).keyBy(0).timeWindow(Time.seconds(60)).apply(new RichWindowFunction<Tuple2<String,Long>, Tuple2<String, Long>, Tuple, TimeWindow>() {
>>
>> @Override
>> public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<Tuple2<String, Long>> collector) throws Exception {
>> log.info("trigger fire at ", System.currentTimeMillis());
>> collector.collect(new Tuple2<>(String.valueOf(timeWindow.toString()), 1l));
>> }
>> }).rebalance().addSink(new FakeSink<>());
>>
>>
>> JobExecutionResult result = env.execute();
>>
>>
>>
>>
>>
>
Re: rocksdb backend on s3 window operator checkpoint issue
Posted by Robert Metzger <rm...@apache.org>.
Hi,
from the code you've provided, everything seems to look okay. I'm currently
trying to reproduce the issue.
Which Flink version are you using?
Which s3 implementation did you configure in the hadoop configuration?
Regards,
Robert
On Mon, May 16, 2016 at 11:52 PM, Chen Qin <qi...@gmail.com> wrote:
> Hi there,
>
> I have been testing checkpointing on rocksdb backed by s3. Checkpoints
> seems successful except snapshot states of timeWindow operator on
> keyedStream. Here is the env setting I used
> env.setStateBackend(new RocksDBStateBackend(new URI("s3://backupdir/")))
>
> The checkpoint for always fail consistently when it goes to window
> operator snapshotting. Exception log attached below.
> I tried to env.setStateBackend(new RocksDBStateBackend(new URI(
> "file:///tmp/checkpoints"))); or MemoryStateBackend(default) works no
> issue with checkpoints.
>
> Does anyone saw this issue before? Or did I mess up with configuration?
>
> Thanks,
> Chen
>
> 2016-05-16 17:20:32,132 INFO
> org.apache.flink.runtime.state.filesystem.FsStateBackend -
> Initializing file state backend to URI
> s3://xxx/checkpoints/7e6abf126ce3d18f173733b34eda81a9
> 2016-05-16 17:20:32,423 INFO
> org.apache.flink.streaming.runtime.tasks.StreamTask - Using
> user-defined state backend:
> org.apache.flink.contrib.streaming.state.RocksDBStateBackend@2fa68a53
> 2016-05-16 17:20:32,423 INFO
> org.apache.flink.runtime.state.filesystem.FsStateBackend -
> Initializing file state backend to URI
> s3://uber-beats/sjc1/checkpoints/7e6abf126ce3d18f173733b34eda81a9
> 2016-05-16 17:21:31,423 INFO
> org.apache.flink.contrib.streaming.state.AbstractRocksDBState - RocksDB
> (/directory/flink-io-723a5c14-2a8a-4abc-881a-9a60138816b0/7e6abf126ce3d18f173733b34eda81a9/WindowOperator_131_0/window-contents/dbc64864de-8373-4b41-bd74-a26a8007f066)
> backup (synchronous part) took 8 ms.
> 2016-05-16 17:21:36,125 ERROR
> org.apache.flink.streaming.runtime.tasks.StreamTask - Caught
> exception while materializing asynchronous checkpoints.
> com.amazonaws.AmazonClientException: Unable to calculate MD5
> hash:/directory//flink-io-723a5c14-2a8a-4abc-881a-9a60138816b0/7e6abf126ce3d18f173733b34eda81a9/WindowOperator_131_0/window-contents/local-chk-599
> (Is a directory)
> at
> com.amazonaws.services.s3.AmazonS3Client.putObject(AmazonS3Client.java:1298)
> at
> com.amazonaws.services.s3.transfer.internal.UploadCallable.uploadInOneChunk(UploadCallable.java:108)
> at
> com.amazonaws.services.s3.transfer.internal.UploadCallable.call(UploadCallable.java:100)
> at
> com.amazonaws.services.s3.transfer.internal.UploadMonitor.upload(UploadMonitor.java:192)
> at
> com.amazonaws.services.s3.transfer.internal.UploadMonitor.call(UploadMonitor.java:150)
> at
> com.amazonaws.services.s3.transfer.internal.UploadMonitor.call(UploadMonitor.java:50)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
>
> ...
>
>
> Tests look like
>
> .setParallelism(1).assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<String>() {
> @Override
> public Watermark checkAndGetNextWatermark(String s, long l) {
> long ts = System.currentTimeMillis() - 60*1000l;
> return new Watermark(ts);
> }
>
> @Override
> public long extractTimestamp(String s, long l) {
> long ts = System.currentTimeMillis();
> return ts;
> }
> }).flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
> @Override
> public void flatMap(String s, Collector<Tuple2<String, Long>> collector) throws Exception {
> collector.collect(new Tuple2<>(s, 1l));
> }
> }).keyBy(0).timeWindow(Time.seconds(60)).apply(new RichWindowFunction<Tuple2<String,Long>, Tuple2<String, Long>, Tuple, TimeWindow>() {
>
> @Override
> public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<Tuple2<String, Long>> iterable, Collector<Tuple2<String, Long>> collector) throws Exception {
> log.info("trigger fire at ", System.currentTimeMillis());
> collector.collect(new Tuple2<>(String.valueOf(timeWindow.toString()), 1l));
> }
> }).rebalance().addSink(new FakeSink<>());
>
>
> JobExecutionResult result = env.execute();
>
>
>
>
>