[HADOOP]数据倾斜的避免和处理

发布时间:2023年12月25日

避免数据倾斜

  1. 初始设计方面

    设计阶段考虑数据分布,并尽可能确保数据均匀分布。
  2. 预处理数据

    在数据加载到 Hadoop 之前进行预处理,以减少倾斜。使用抽样或统计方法来了解数据分布特征,并据此调整。
  3. 使用合适的Partitioner

    自定义Partitioner,替换Hadoop默认的hash-based partitioner,来更均匀地分配数据。
  4. 增加Reduce任务数量

    通过设置更多的Reduce任务来帮助将工作更细致地分散开。
  5. 利用Combiner

    在Map阶段尽早使用Combiner来减少需要传输到Reducer的数据量。这个方法需要关注Combiner对reducer数量的变化是否造成业务结果的不准确。
  6. Bucketing

    Hive中的bucketing可以帮助预先对数据进行分区和排序,从而缓解倾斜。

处理现有的数据倾斜

  1. 采样并动态调整

    对输入数据进行采样,根据采样结果动态调整partitioner的逻辑。在运行MapReduce任务之前,基于采样数据创建一个分布式Cache,使自定义partitioner可以使用这些信息来决定数据如何被送到Reducers。
  2. 再平衡负载

    在检测到数据倾斜后,手动或者程序性地进行工作重新分配。为处理大量数据的Reducers添加更多资源。
  3. 拆分大键

    如果数据倾斜是因为某个特定“热点”键值造成的,尝试将该键分成多个键。
  4. 采用Salting技术

    对键值进行"Salting",即给键添加随机前缀,从而改变数据的分布。在Reducer端做相应的聚合处理。
  5. 使用Skew-Join优化策略

    如果倾斜出现在Join操作中,使用诸如Map-side join、Reduce-side join 或 Skew-Join等策略来优化。
  6. 限制Reducer输入

    为Reducer设置大小阈值,限制处理过多数据。
  7. 使用外部工具

    利用Apache Tez或Spark等更高级的数据处理框架可以提供更好的控制来减少数据倾斜问题。

避免数据倾斜(附上代码)

1. 使用合适的Partitioner:

如果某个键特别频繁,它可能会导致一个Reducer负载过重。可以自定义Partitioner来分散这种热点键。

public class CustomPartitioner extends Partitioner<Text, IntWritable> {
    @Override
    public int getPartition(Text key, IntWritable value, int numReduceTasks) {
        if(key.toString().equals("hotkey")) {
            return 0; 
        }
        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
}

2.增加Reduce任务数量:

有时候简单的增加Reducers的数量也能缓解数据倾斜。

hadoop jar your-job.jar YourDriverClass -Dmapreduce.job.reduces=100 inputPath outputPath

处理现有的数据倾斜(附上代码)

1.采样并动态调整:

假设你有大量数据聚集在某些键上,通过对数据进行采样,我们可以确定如何更好地划分数据。

在MapReduce的setup阶段读取采样数据,并根据这些数据来确定分区逻辑。

public class SkewAwareMapper extends Mapper<...> {
    private HashMap<String, Integer> distributionMap;

    protected void setup(Context context) throws IOException, InterruptedException {
        distributionMap = ... // Load the sampled distribution map
    }

    protected void map(... key, ... value, Context context) throws IOException, InterruptedException {
        // Use the distribution map to decide how to spread out keys
        ...
    }
}

2.拆分大键:

当一个键异常大时,可以将该键拆分成多个子键。

public class LargeKeySplittingMapper extends Mapper<LongWritable, Text, Text, Text> {
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String originalKey = value.toString().split("\t")[0];
        String payload = value.toString().split("\t")[1];

        // Implementing key splitting logic
        if (originalKey.equals("massiveKey")) {
            // Split the massive key into smaller ones
            for (int i = 0; i < 10; i++) {
                context.write(new Text(originalKey + "_part_" + i), new Text(payload));
            }
        } else {
            context.write(new Text(originalKey), new Text(payload));
        }
    }
}

3.Salting技术:

如果你知道某些键出现频率很高,你可以在写入数据时为这些键添加随机前缀(salting),以均匀分配。

public class SaltingMapper extends Mapper<LongWritable, Text, Text, Text> {
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String originalKey = value.toString().split("\t")[0];
        String payload = value.toString().split("\t")[1];

        Random rand = new Random();
        int salt = rand.nextInt(10); // Create 10 partitions for each key
        
        context.write(new Text(salt + "_" + originalKey), new Text(payload));
    }
}

文章来源:https://blog.csdn.net/weixin_46485607/article/details/135194490
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。