Skip to content

[HUDI-3438] Avoid getSmallFiles if hoodie.parquet.small.file.limit is 0 #4823

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nsivabalan merged 3 commits into from
Feb 18, 2022
Merged

[HUDI-3438] Avoid getSmallFiles if hoodie.parquet.small.file.limit is 0 #4823

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0557d89
Avoid getSmallFiles if hoodie.parquet.small.file.limit is 0
Feb 16, 2022
27f65c3
Fix NPE
Feb 16, 2022
85eb9a4
Add document
Feb 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion ...-java-client/src/main/java/org/apache/hudi/table/action/commit/JavaUpsertPartitioner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,7 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context)
WorkloadStat pStat = profile.getWorkloadStat(partitionPath);
if (pStat.getNumInserts() > 0) {

List<SmallFile> smallFiles = partitionSmallFilesMap.get(partitionPath);
List<SmallFile> smallFiles = partitionSmallFilesMap.getOrDefault(partitionPath, new ArrayList<>());
this.smallFiles.addAll(smallFiles);

LOG.info("For partitionPath : " + partitionPath + " Small Files => " + smallFiles);
Expand Down Expand Up @@ -205,6 +205,11 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context)

private Map<String, List<SmallFile>> getSmallFilesForPartitions(List<String> partitionPaths, HoodieEngineContext context) {
Map<String, List<SmallFile>> partitionSmallFilesMap = new HashMap<>();

if (config.getParquetSmallFileLimit() == 0) {

@YuweiXiao YuweiXiao Feb 16, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about using <= 0 to indicate always writing to new file group? And maybe add a description of this behavior to the document of PARQUET_SMALL_FILE_LIMIT.

@boneanxs boneanxs Feb 17, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can PARQUET_SMALL_FILE_LIMIT be set to be negative? I'm not sure.

Yeah, I'll add the description to the PARQUET_SMALL_FILE_LIMIT

@YuweiXiao YuweiXiao Feb 17, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is possible to be set it to negative, which is meaningless of course. So I guess we could cover it as well?

@boneanxs boneanxs Feb 18, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

done

return partitionSmallFilesMap;
}

if (partitionPaths != null && partitionPaths.size() > 0) {
context.setJobStatus(this.getClass().getSimpleName(), "Getting small files from partitions");
partitionSmallFilesMap = context.mapToPair(partitionPaths,
Expand Down
7 changes: 6 additions & 1 deletion ...udi-spark-client/src/main/java/org/apache/hudi/table/action/commit/UpsertPartitioner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,7 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context)

List<SmallFile> smallFiles =
filterSmallFilesInClustering(partitionPathToPendingClusteringFileGroupsId.getOrDefault(partitionPath, Collections.emptySet()),
partitionSmallFilesMap.get(partitionPath));
partitionSmallFilesMap.getOrDefault(partitionPath, new ArrayList<>()));

this.smallFiles.addAll(smallFiles);

Expand Down Expand Up @@ -241,6 +241,11 @@ private void assignInserts(WorkloadProfile profile, HoodieEngineContext context)
private Map<String, List<SmallFile>> getSmallFilesForPartitions(List<String> partitionPaths, HoodieEngineContext context) {
JavaSparkContext jsc = HoodieSparkEngineContext.getSparkContext(context);
Map<String, List<SmallFile>> 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<String> partitionPathRdds = jsc.parallelize(partitionPaths, partitionPaths.size());
Expand Down
4 changes: 4 additions & 0 deletions ...main/java/org/apache/hudi/table/action/deltacommit/SparkUpsertDeltaCommitPartitioner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,10 @@ private List<FileSlice> 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()
Expand Down
9 changes: 8 additions & 1 deletion hudi-flink/src/main/java/org/apache/hudi/sink/partitioner/profile/WriteProfile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,14 @@ public synchronized List<SmallFile> getSmallFiles(String partitionPath) {
if (smallFilesMap.containsKey(partitionPath)) {
return smallFilesMap.get(partitionPath);
}
List<SmallFile> smallFiles = smallFilesProfile(partitionPath);

List<SmallFile> smallFiles = new ArrayList<>();
if (config.getParquetSmallFileLimit() == 0) {
this.smallFilesMap.put(partitionPath, smallFiles);
return smallFiles;
}

smallFiles = smallFilesProfile(partitionPath);
this.smallFilesMap.put(partitionPath, smallFiles);
return smallFiles;
}
Expand Down