You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by 耿延杰 <gy...@qq.com> on 2021/01/27 08:21:50 UTC
回复：Setting "unreliable" RocksDB state backend w/o "execution.checkpointing.interval" and "state.checkpoints.dir" properties
Hi,
As far as I know, Flink init RocksDBStateBackend.class Object by
- your user-code config
RocksDBStateBackend rocksDBStateBackend =
new RocksDBStateBackend(EventProcessorConstant.CHECKPOINT_PATH, true);
env.setStateBackend(rocksDBStateBackend);
- if you don't set in your code, then by config file
When we look deep into RocksDBStateBackend.class[1]
All of it's constructor methods require 'CheckpointPathUri'.
Thus, you must set checkpoint path in config file, even you disable checkpoint.
[1]:https://ci.apache.org/projects/flink/flink-docs-stable/api/java/index.html?org/apache/flink/contrib/streaming/state/RocksDBStateBackend.html
------------------ 原始邮件 ------------------
发件人: "Dongwon Kim" <eastcirclek@gmail.com>;
发送时间: 2021年1月27日(星期三) 下午3:18
收件人: "user"<user@flink.apache.org>;
主题: Setting "unreliable" RocksDB state backend w/o "execution.checkpointing.interval" and "state.checkpoints.dir" properties
Hi,
I've defined some states in an instance of KeyedProcessFunction so that each key has its own state.
As the size of the stats can be quite large, I want them to be stored on local disks via a RocksDB state backend.
To that end, I specified only a single property in my flink-conf.yaml:
state.backend: rocksdb
However, it causes the following exception:
Caused by: org.apache.flink.configuration.IllegalConfigurationException: Cannot create the RocksDB state backend: The configuration does not specify the checkpoint directory 'state.checkpoints.dir'
at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:45)
at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:32)
at org.apache.flink.runtime.state.StateBackendLoader.loadStateBackendFromConfig(StateBackendLoader.java:165)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.loadStateBackend(StreamExecutionEnvironment.java:854)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.configure(StreamExecutionEnvironment.java:797)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:243)
at org.apache.flink.client.program.StreamContextEnvironment.<init>(StreamContextEnvironment.java:67)
at org.apache.flink.client.program.StreamContextEnvironment.lambda$setAsContext$4(StreamContextEnvironment.java:165)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.lambda$getExecutionEnvironment$12(StreamExecutionEnvironment.java:2051)
at java.util.Optional.map(Optional.java:215)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:2051)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:2034)
at com.kakaomobility.drivinghabit.stream.BlueGreenComparator.main(BlueGreenComparator.java:120)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:343)
... 11 more
Okay, let's add state.checkpoints.dir in flink-conf.yaml:
state.backend: rocksdb
state.checkpoints.dir: hdfs:///path/to/ckpts
But the periodic checkpoints are essentially disabled as I didn't specify "execution.checkpointing.interval" and it's also shown in the attached screenshot.
What if I want to use RocksDB as an unreliable per-key state storage?
Do I still have to specify "state.checkpoints.dir" which does totally nothing here?
Best,
Dongwon
回复： Setting "unreliable" RocksDB state backend w/o "execution.checkpointing.interval" and "state.checkpoints.dir" properties
Posted by 耿延杰 <gy...@qq.com>.
Hi Dongwon,
Involve Congxian Qiu who is expert in Flink.
Hi Congxian Qiu,
Would you like to answer Dongwon's question and
what do you think about his suggestion?
------------------ 原始邮件 ------------------
发件人: "Dongwon Kim" <eastcirclek@gmail.com>;
发送时间: 2021年1月28日(星期四) 上午8:25
收件人: "耿延杰"<gyj199482@qq.com>;
抄送: "user"<user@flink.apache.org>;
主题: Re: Setting "unreliable" RocksDB state backend w/o "execution.checkpointing.interval" and "state.checkpoints.dir" properties
Hi 耿延杰,
Would it be better to require "state.checkpoints.dir" only when "execution.checkpointing.interval" is not none?
Best,
Dongwon
On Wed, Jan 27, 2021 at 5:22 PM 耿延杰 <gyj199482@qq.com> wrote:
Hi,
As far as I know, Flink init RocksDBStateBackend.class Object by
- your user-code config
RocksDBStateBackend rocksDBStateBackend =
new RocksDBStateBackend(EventProcessorConstant.CHECKPOINT_PATH, true);
env.setStateBackend(rocksDBStateBackend);
- if you don't set in your code, then by config file
When we look deep into RocksDBStateBackend.class[1]
All of it's constructor methods require 'CheckpointPathUri'.
Thus, you must set checkpoint path in config file, even you disable checkpoint.
[1]:https://ci.apache.org/projects/flink/flink-docs-stable/api/java/index.html?org/apache/flink/contrib/streaming/state/RocksDBStateBackend.html
------------------ 原始邮件 ------------------
发件人: "Dongwon Kim" <eastcirclek@gmail.com>;
发送时间: 2021年1月27日(星期三) 下午3:18
收件人: "user"<user@flink.apache.org>;
主题: Setting "unreliable" RocksDB state backend w/o "execution.checkpointing.interval" and "state.checkpoints.dir" properties
Hi,
I've defined some states in an instance of KeyedProcessFunction so that each key has its own state.
As the size of the stats can be quite large, I want them to be stored on local disks via a RocksDB state backend.
To that end, I specified only a single property in my flink-conf.yaml:
state.backend: rocksdb
However, it causes the following exception:
Caused by: org.apache.flink.configuration.IllegalConfigurationException: Cannot create the RocksDB state backend: The configuration does not specify the checkpoint directory 'state.checkpoints.dir'
at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:45)
at org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:32)
at org.apache.flink.runtime.state.StateBackendLoader.loadStateBackendFromConfig(StateBackendLoader.java:165)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.loadStateBackend(StreamExecutionEnvironment.java:854)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.configure(StreamExecutionEnvironment.java:797)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:243)
at org.apache.flink.client.program.StreamContextEnvironment.<init>(StreamContextEnvironment.java:67)
at org.apache.flink.client.program.StreamContextEnvironment.lambda$setAsContext$4(StreamContextEnvironment.java:165)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.lambda$getExecutionEnvironment$12(StreamExecutionEnvironment.java:2051)
at java.util.Optional.map(Optional.java:215)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:2051)
at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:2034)
at com.kakaomobility.drivinghabit.stream.BlueGreenComparator.main(BlueGreenComparator.java:120)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:343)
... 11 more
Okay, let's add state.checkpoints.dir in flink-conf.yaml:
state.backend: rocksdb
state.checkpoints.dir: hdfs:///path/to/ckpts
But the periodic checkpoints are essentially disabled as I didn't specify "execution.checkpointing.interval" and it's also shown in the attached screenshot.
What if I want to use RocksDB as an unreliable per-key state storage?
Do I still have to specify "state.checkpoints.dir" which does totally nothing here?
Best,
Dongwon
Re: Setting "unreliable" RocksDB state backend w/o
"execution.checkpointing.interval" and "state.checkpoints.dir" properties
Posted by Dongwon Kim <ea...@gmail.com>.
Hi 耿延杰,
Would it be better to require "state.checkpoints.dir" only when
"execution.checkpointing.interval" is not none?
Best,
Dongwon
On Wed, Jan 27, 2021 at 5:22 PM 耿延杰 <gy...@qq.com> wrote:
> Hi,
>
> As far as I know, Flink init RocksDBStateBackend.class Object by
> - your user-code config
>
> RocksDBStateBackend rocksDBStateBackend =
> new RocksDBStateBackend(EventProcessorConstant.CHECKPOINT_PATH, true);
> env.setStateBackend(rocksDBStateBackend);
>
> - if you don't set in your code, then by config file
>
> When we look deep into RocksDBStateBackend.class[1]
> All of it's constructor methods require 'CheckpointPathUri'.
>
> Thus, you must set checkpoint path in config file, even you disable
> checkpoint.
>
>
> [1]:
> https://ci.apache.org/projects/flink/flink-docs-stable/api/java/index.html?org/apache/flink/contrib/streaming/state/RocksDBStateBackend.html
>
>
>
> ------------------ 原始邮件 ------------------
> *发件人:* "Dongwon Kim" <ea...@gmail.com>;
> *发送时间:* 2021年1月27日(星期三) 下午3:18
> *收件人:* "user"<us...@flink.apache.org>;
> *主题:* Setting "unreliable" RocksDB state backend w/o
> "execution.checkpointing.interval" and "state.checkpoints.dir" properties
>
> Hi,
>
> I've defined some states in an instance of KeyedProcessFunction so that
> each key has its own state.
> As the size of the stats can be quite large, I want them to be stored on
> local disks via a RocksDB state backend.
> To that end, I specified only a single property in my flink-conf.yaml:
>
>> state.backend: rocksdb
>
>
> However, it causes the following exception:
>
>> Caused by: org.apache.flink.configuration.IllegalConfigurationException:
>> Cannot create the RocksDB state backend: The configuration does not specify
>> the checkpoint directory 'state.checkpoints.dir'
>> at
>> org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:45)
>> at
>> org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory.createFromConfig(RocksDBStateBackendFactory.java:32)
>> at
>> org.apache.flink.runtime.state.StateBackendLoader.loadStateBackendFromConfig(StateBackendLoader.java:165)
>> at
>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.loadStateBackend(StreamExecutionEnvironment.java:854)
>> at
>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.configure(StreamExecutionEnvironment.java:797)
>> at
>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.<init>(StreamExecutionEnvironment.java:243)
>> at
>> org.apache.flink.client.program.StreamContextEnvironment.<init>(StreamContextEnvironment.java:67)
>> at
>> org.apache.flink.client.program.StreamContextEnvironment.lambda$setAsContext$4(StreamContextEnvironment.java:165)
>> at
>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.lambda$getExecutionEnvironment$12(StreamExecutionEnvironment.java:2051)
>> at java.util.Optional.map(Optional.java:215)
>> at
>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:2051)
>> at
>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(StreamExecutionEnvironment.java:2034)
>> at
>> com.kakaomobility.drivinghabit.stream.BlueGreenComparator.main(BlueGreenComparator.java:120)
>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>> at
>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>> at java.lang.reflect.Method.invoke(Method.java:498)
>> at
>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:343)
>> ... 11 more
>
>
> Okay, let's add state.checkpoints.dir in flink-conf.yaml:
>
>> state.backend: rocksdb
>
> state.checkpoints.dir: hdfs:///path/to/ckpts
>
>
> But the periodic checkpoints are essentially disabled as I didn't specify
> "execution.checkpointing.interval" and it's also shown in the attached
> screenshot.
>
>
> What if I want to use RocksDB as an unreliable per-key state storage?
> Do I still have to specify "state.checkpoints.dir" which does totally
> nothing here?
>
> Best,
>
> Dongwon
>
>