Hadoop clusters can suffer from performance degradasion when data is unevenly distributed across nodes. This imbalance leads to some node being overloaded while others remain idle. The MapReduce paradigm splits data into blocks for parallel processing, but if block sizes or distribution are skewed, the overall job efficiency drops. A practical solution is to implement a custom InputFormat that dynamically adjusts input splits for more uniform distribution.

Below is an example implementation that extends TextInputFormat and overrides the getSplits method. It calculates the average split size across all blocks and iteratively splits any block exceeding that average, generating smaller, more balanced splits.

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class BalancedInputFormat extends TextInputFormat {
    @Override
    public List<InputSplit> getSplits(JobContext context) throws IOException {
        List<InputSplit> originalSplits = super.getSplits(context);
        
        // Compute total size of all splits
        long totalSize = 0;
        for (InputSplit split : originalSplits) {
            totalSize += split.getLength();
        }
        long averageSize = totalSize / originalSplits.size();
        
        List<InputSplit> balancedSplits = new ArrayList<>();
        for (InputSplit split : originalSplits) {
            long remaining = split.getLength();
            long offset = split.getStart();
            while (remaining > averageSize) {
                balancedSplits.add(new FileSplit(
                    split.getPath(), offset, averageSize, split.getLocations()));
                offset += averageSize;
                remaining -= averageSize;
            }
            // Add the final remainder
            if (remaining > 0) {
                balancedSplits.add(new FileSplit(
                    split.getPath(), offset, remaining, split.getLocations()));
            }
        }
        return balancedSplits;
    }
}

This approach avoids a manual repartitioning of HDFS data and works at the job level. It ensures that no single split is significantly larger than the average, reducing the risk of straggler tasks. For production use, you may also need to consider factors like data locality and ongoing data changes, but this provides a solid starting point for improving balance.