lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-1812] Add explicit index state TTL option for Flink writer (#2853)

Browse Source
This commit is contained in:
hiscat
2021-04-21 20:13:30 +08:00
committed by GitHub
parent ac3589f006
commit cc81ddde01
2 changed files with 12 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
hudi-flink/src/main/java/org/apache/hudi/configuration/FlinkOptions.java
View File

@@ -80,6 +80,12 @@ public class FlinkOptions {
.defaultValue(false)
.withDescription("Whether to bootstrap the index state from existing hoodie table, default false");
public static final ConfigOption<Double> INDEX_STATE_TTL = ConfigOptions
.key("index.state.ttl")
.doubleType()
.defaultValue(1.5D)
.withDescription("Index state ttl in days, default 1.5 day");
// ------------------------------------------------------------------------
// Read Options
// ------------------------------------------------------------------------

7
hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/BucketAssignFunction.java
View File

@@ -37,17 +37,18 @@ import org.apache.hudi.table.action.commit.BucketInfo;
import org.apache.hudi.util.StreamerUtil;
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.CheckpointListener;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.table.runtime.util.StateTtlConfigUtil;
import org.apache.flink.util.Collector;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
@@ -145,6 +146,10 @@ public class BucketAssignFunction<K, I, O extends HoodieRecord<?>>
"indexState",
TypeInformation.of(HoodieKey.class),
TypeInformation.of(HoodieRecordLocation.class));
double ttl = conf.getDouble(FlinkOptions.INDEX_STATE_TTL) * 24 * 60 * 60 * 1000;
if (ttl > 0) {
indexStateDesc.enableTimeToLive(StateTtlConfigUtil.createTtlConfig((long) ttl));
}
indexState = context.getKeyedStateStore().getMapState(indexStateDesc);
if (bootstrapIndex) {
MapStateDescriptor<String, Integer> partitionLoadStateDesc =