lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-4366] Synchronous cleaning for flink bounded source (#6051)

Browse Source
This commit is contained in:
Danny Chan
2022-07-08 09:55:07 +08:00
committed by GitHub
parent 5673819736
commit c744848c59
2 changed files with 10 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/configuration/OptionsResolver.java
View File

@@ -164,4 +164,12 @@ public class OptionsResolver {
public static boolean sortClusteringEnabled(Configuration conf) { public static boolean sortClusteringEnabled(Configuration conf) {
return !StringUtils.isNullOrEmpty(conf.getString(FlinkOptions.CLUSTERING_SORT_COLUMNS)); return !StringUtils.isNullOrEmpty(conf.getString(FlinkOptions.CLUSTERING_SORT_COLUMNS));
} }
/**
* Returns whether the operation is INSERT OVERWRITE (table or partition).
*/
public static boolean isInsertOverwrite(Configuration conf) {
return conf.getString(FlinkOptions.OPERATION).equals(WriteOperationType.INSERT_OVERWRITE_TABLE.value())
|| conf.getString(FlinkOptions.OPERATION).equals(WriteOperationType.INSERT_OVERWRITE.value());
}
} }

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

@@ -19,9 +19,9 @@
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.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.configuration.FlinkOptions;
import org.apache.hudi.configuration.OptionsResolver;
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;
@@ -65,7 +65,7 @@ public class CleanFunction<T> extends AbstractRichFunction
this.writeClient = StreamerUtil.createWriteClient(conf, getRuntimeContext()); this.writeClient = StreamerUtil.createWriteClient(conf, getRuntimeContext());
this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build(); this.executor = NonThrownExecutor.builder(LOG).waitForTasksFinish(true).build();
if (conf.getString(FlinkOptions.OPERATION).equals(WriteOperationType.INSERT_OVERWRITE_TABLE.value())) { if (OptionsResolver.isInsertOverwrite(conf)) {
String instantTime = HoodieActiveTimeline.createNewInstantTime(); String instantTime = HoodieActiveTimeline.createNewInstantTime();
LOG.info(String.format("exec sync clean with instant time %s...", instantTime)); LOG.info(String.format("exec sync clean with instant time %s...", instantTime));
executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish"); executor.execute(() -> writeClient.clean(instantTime), "wait for sync cleaning finish");