Skip to content

Reduce the maximum size of input splits in Flink to better distribute work #28045

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
robertwb merged 4 commits into from
Sep 1, 2023
Merged

Reduce the maximum size of input splits in Flink to better distribute work #28045

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4e8acc0
Flink input split size change
jto Aug 11, 2023
40d2093
Add Flink split size option
jto Aug 17, 2023
9579fe0
[Flink] Set default max input split size to -1 (disabled)
jto Aug 28, 2023
19f9423
Fix NPE in tests
jto Aug 28, 2023
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: 7 additions & 0 deletions runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkPipelineOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -313,6 +313,13 @@ public interface FlinkPipelineOptions

void setFlinkConfDir(String confDir);

@Description(
"Set the maximum size of input split when data is read from a filesystem. 0 implies no max size.")
@Default.Long(0)
Long getFileInputSplitMaxSizeMB();

void setFileInputSplitMaxSizeMB(Long fileInputSplitMaxSizeMB);

static FlinkPipelineOptions defaults() {
return PipelineOptionsFactory.as(FlinkPipelineOptions.class);
}
Expand Down
22 changes: 21 additions & 1 deletion ...k/src/main/java/org/apache/beam/runners/flink/translation/wrappers/SourceInputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,11 @@
import java.io.IOException;
import java.util.List;
import org.apache.beam.runners.core.construction.SerializablePipelineOptions;
import org.apache.beam.runners.flink.FlinkPipelineOptions;
import org.apache.beam.runners.flink.metrics.FlinkMetricContainer;
import org.apache.beam.runners.flink.metrics.ReaderInvocationUtil;
import org.apache.beam.sdk.io.BoundedSource;
import org.apache.beam.sdk.io.FileBasedSource;
import org.apache.beam.sdk.io.Source;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
Expand Down Expand Up @@ -108,12 +110,30 @@ public float getAverageRecordWidth() {
return null;
}

private long getDesiredSizeBytes(int numSplits) throws Exception {
long totalSize = initialSource.getEstimatedSizeBytes(options);
long defaultSplitSize = totalSize / numSplits;
long maxSplitSize = 0;
if (options != null) {
maxSplitSize = options.as(FlinkPipelineOptions.class).getFileInputSplitMaxSizeMB();
}
if (initialSource instanceof FileBasedSource && maxSplitSize > 0) {
// Most of the time parallelism is < number of files in source.
// Each file becomes a unique split which commonly create skew.
// This limits the size of splits to reduce skew.
return Math.min(defaultSplitSize, maxSplitSize * 1024 * 1024);
} else {
return defaultSplitSize;
}
}

@Override
@SuppressWarnings("unchecked")
public SourceInputSplit<T>[] createInputSplits(int numSplits) throws IOException {
try {
long desiredSizeBytes = initialSource.getEstimatedSizeBytes(options) / numSplits;
long desiredSizeBytes = getDesiredSizeBytes(numSplits);
List<? extends Source<T>> shards = initialSource.split(desiredSizeBytes, options);

int numShards = shards.size();
SourceInputSplit<T>[] sourceInputSplits = new SourceInputSplit[numShards];
for (int i = 0; i < numShards; i++) {
Expand Down