From 0557d89f16796f6dd1602dabc723f5f8367b177a Mon Sep 17 00:00:00 2001 From: Hui An Date: Wed, 16 Feb 2022 11:55:08 +0800 Subject: [PATCH 1/3] Avoid getSmallFiles if hoodie.parquet.small.file.limit is 0 --- .../hudi/table/action/commit/JavaUpsertPartitioner.java | 5 +++++ .../hudi/table/action/commit/UpsertPartitioner.java | 5 +++++ .../deltacommit/SparkUpsertDeltaCommitPartitioner.java | 4 ++++ .../hudi/sink/partitioner/profile/WriteProfile.java | 9 ++++++++- 4 files changed, 22 insertions(+), 1 deletion(-) diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java index 33f59f4406f39..3da59fd0a3724 100644 --- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java +++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java @@ -205,6 +205,11 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context) private Map> getSmallFilesForPartitions(List partitionPaths, HoodieEngineContext context) { Map> partitionSmallFilesMap = new HashMap<>(); + + if (config.getParquetSmallFileLimit() == 0) { + return partitionSmallFilesMap; + } + if (partitionPaths != null && partitionPaths.size() > 0) { context.setJobStatus(this.getClass().getSimpleName(), "Getting small files from partitions"); partitionSmallFilesMap = context.mapToPair(partitionPaths, diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java index 6729da72d65eb..d7c997388522d 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java @@ -241,6 +241,11 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context) private Map> getSmallFilesForPartitions(List partitionPaths, HoodieEngineContext context) { JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(context); Map> partitionSmallFilesMap = new HashMap<>(); + + if (config.getParquetSmallFileLimit() == 0) { + return partitionSmallFilesMap; + } + if (partitionPaths != null && partitionPaths.size() > 0) { context.setJobStatus(this.getClass().getSimpleName(), "Getting small files from partitions"); JavaRDD partitionPathRdds = jsc.parallelize(partitionPaths, partitionPaths.size()); diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java index 8dd3146f5161d..ddc01a868b789 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java @@ -97,6 +97,10 @@ private List getSmallFileCandidates(String partitionPath, HoodieInsta .collect(Collectors.toList()); } + if (config.getParquetSmallFileLimit() == 0) { + return Collections.emptyList(); + } + // If we cannot index log files, then we choose the smallest parquet file in the partition and add inserts to // it. Doing this overtime for a partition, we ensure that we handle small file issues return table.getSliceView() diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java index 84fcd03f0833b..b5a87048d8cae 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java @@ -180,7 +180,14 @@ public synchronized List getSmallFiles(String partitionPath) { if (smallFilesMap.containsKey(partitionPath)) { return smallFilesMap.get(partitionPath); } - List smallFiles = smallFilesProfile(partitionPath); + + List smallFiles = new ArrayList<>(); + if (config.getParquetSmallFileLimit() == 0) { + this.smallFilesMap.put(partitionPath, smallFiles); + return smallFiles; + } + + smallFiles = smallFilesProfile(partitionPath); this.smallFilesMap.put(partitionPath, smallFiles); return smallFiles; } From 27f65c3f50cb2a2048ddc07ff7848d4d46fb9fa3 Mon Sep 17 00:00:00 2001 From: Hui An Date: Wed, 16 Feb 2022 12:53:18 +0800 Subject: [PATCH 2/3] Fix NPE --- .../apache/hudi/table/action/commit/JavaUpsertPartitioner.java | 2 +- .../org/apache/hudi/table/action/commit/UpsertPartitioner.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java index 3da59fd0a3724..700a34356d096 100644 --- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java +++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java @@ -135,7 +135,7 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context) WorkloadStat pStat = profile.getWorkloadStat(partitionPath); if (pStat.getNumInserts() > 0) { - List smallFiles = partitionSmallFilesMap.get(partitionPath); + List smallFiles = partitionSmallFilesMap.getOrDefault(partitionPath, new ArrayList<>()); this.smallFiles.addAll(smallFiles); LOG.info("For partitionPath : " + partitionPath + " Small Files => " + smallFiles); diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java index d7c997388522d..88eb53a452d61 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java @@ -165,7 +165,7 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context) List smallFiles = filterSmallFilesInClustering(partitionPathToPendingClusteringFileGroupsId.getOrDefault(partitionPath, Collections.emptySet()), - partitionSmallFilesMap.get(partitionPath)); + partitionSmallFilesMap.getOrDefault(partitionPath, new ArrayList<>())); this.smallFiles.addAll(smallFiles); From 85eb9a4c306611fee5309afe9404aa82da12fd7f Mon Sep 17 00:00:00 2001 From: Hui An Date: Fri, 18 Feb 2022 10:14:39 +0800 Subject: [PATCH 3/3] Add document --- .../java/org/apache/hudi/config/HoodieCompactionConfig.java | 3 ++- .../apache/hudi/table/action/commit/JavaUpsertPartitioner.java | 2 +- .../org/apache/hudi/table/action/commit/UpsertPartitioner.java | 2 +- .../action/deltacommit/SparkUpsertDeltaCommitPartitioner.java | 2 +- .../org/apache/hudi/sink/partitioner/profile/WriteProfile.java | 2 +- 5 files changed, 6 insertions(+), 5 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java index 0d0984354067f..65b36fcb62233 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieCompactionConfig.java @@ -173,7 +173,8 @@ public class HoodieCompactionConfig extends HoodieConfig { .defaultValue(String.valueOf(104857600)) .withDocumentation("During upsert operation, we opportunistically expand existing small files on storage, instead of writing" + " new files, to keep number of files to an optimum. This config sets the file size limit below which a file on storage " - + " becomes a candidate to be selected as such a `small file`. By default, treat any file <= 100MB as a small file."); + + " becomes a candidate to be selected as such a `small file`. By default, treat any file <= 100MB as a small file." + + " Also note that if this set <= 0, will not try to get small files and directly write new files"); public static final ConfigProperty RECORD_SIZE_ESTIMATION_THRESHOLD = ConfigProperty .key("hoodie.record.size.estimation.threshold") diff --git a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java index 700a34356d096..f8aaba52934e2 100644 --- a/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java +++ b/hudi-client/hudi-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java @@ -206,7 +206,7 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context) private Map> getSmallFilesForPartitions(List partitionPaths, HoodieEngineContext context) { Map> partitionSmallFilesMap = new HashMap<>(); - if (config.getParquetSmallFileLimit() == 0) { + if (config.getParquetSmallFileLimit() <= 0) { return partitionSmallFilesMap; } diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java index 88eb53a452d61..98f2539c69f38 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java @@ -242,7 +242,7 @@ private Map> getSmallFilesForPartitions(List par JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(context); Map> partitionSmallFilesMap = new HashMap<>(); - if (config.getParquetSmallFileLimit() == 0) { + if (config.getParquetSmallFileLimit() <= 0) { return partitionSmallFilesMap; } diff --git a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java index ddc01a868b789..e498019c415d8 100644 --- a/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java +++ b/hudi-client/hudi-spark-client/src/main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java @@ -97,7 +97,7 @@ private List getSmallFileCandidates(String partitionPath, HoodieInsta .collect(Collectors.toList()); } - if (config.getParquetSmallFileLimit() == 0) { + if (config.getParquetSmallFileLimit() <= 0) { return Collections.emptyList(); } diff --git a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java index b5a87048d8cae..fdb8152b00577 100644 --- a/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java +++ b/hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java @@ -182,7 +182,7 @@ public synchronized List getSmallFiles(String partitionPath) { } List smallFiles = new ArrayList<>(); - if (config.getParquetSmallFileLimit() == 0) { + if (config.getParquetSmallFileLimit() <= 0) { this.smallFilesMap.put(partitionPath, smallFiles); return smallFiles; }