lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-3573] flink cleanFuntion execute clean on initialization (#4936)

Browse Source 
For flink insert overwrite operation, do the cleaning each time before the write.
This commit is contained in:
todd5167
2022-03-08 11:53:54 +08:00
committed by GitHub
parent 29040762fa
commit 34bc752853
1 changed files with 13 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
hudi-flink/src/main/java/org/apache/hudi/sink/CleanFunction.java
View File

@@ -19,6 +19,8 @@
package org.apache.hudi.sink; package org.apache.hudi.sink;
import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.client.HoodieFlinkWriteClient;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.sink.utils.NonThrownExecutor; import org.apache.hudi.sink.utils.NonThrownExecutor;
import org.apache.hudi.util.StreamerUtil; import org.apache.hudi.util.StreamerUtil;
@@ -63,7 +65,13 @@ public class CleanFunction<T> extends AbstractRichFunction
// do not use the remote filesystem view because the async cleaning service // do not use the remote filesystem view because the async cleaning service
// local timeline is very probably to fall behind with the remote one. // local timeline is very probably to fall behind with the remote one.
this.writeClient = StreamerUtil.createWriteClient(conf, getRuntimeContext(), false); this.writeClient = StreamerUtil.createWriteClient(conf, getRuntimeContext(), false);
this.executor = NonThrownExecutor.builder(LOG).build(); this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (conf.getString(FlinkOptions.OPERATION).equals(WriteOperationType.INSERT_OVERWRITE_TABLE.value())) {
String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");
}
} }
} }
@@ -101,6 +109,10 @@ public class CleanFunction<T> extends AbstractRichFunction
@Override @Override
public void close() throws Exception { public void close() throws Exception {
if (executor != null) {
executor.close();
}
if (this.writeClient != null) { if (this.writeClient != null) {
this.writeClient.close(); this.writeClient.close();
} }