避免数据倾斜
-
初始设计方面：
设计阶段考虑数据分布，并尽可能确保数据均匀分布。 -
预处理数据：
在数据加载到 Hadoop 之前进行预处理，以减少倾斜。使用抽样或统计方法来了解数据分布特征，并据此调整。 -
使用合适的Partitioner：
自定义Partitioner，替换Hadoop默认的hash-based partitioner，来更均匀地分配数据。 -
增加Reduce任务数量：
通过设置更多的Reduce任务来帮助将工作更细致地分散开。 -
利用Combiner：
在Map阶段尽早使用Combiner来减少需要传输到Reducer的数据量。这个方法需要关注Combiner对reducer数量的变化是否造成业务结果的不准确。 -
Bucketing：
Hive中的bucketing可以帮助预先对数据进行分区和排序，从而缓解倾斜。
处理现有的数据倾斜
-
采样并动态调整：
对输入数据进行采样，根据采样结果动态调整partitioner的逻辑。在运行MapReduce任务之前，基于采样数据创建一个分布式Cache，使自定义partitioner可以使用这些信息来决定数据如何被送到Reducers。 -
再平衡负载：
在检测到数据倾斜后，手动或者程序性地进行工作重新分配。为处理大量数据的Reducers添加更多资源。 -
拆分大键：
如果数据倾斜是因为某个特定“热点”键值造成的，尝试将该键分成多个键。 -
采用Salting技术：
对键值进行"Salting"，即给键添加随机前缀，从而改变数据的分布。在Reducer端做相应的聚合处理。 -
使用Skew-Join优化策略：
如果倾斜出现在Join操作中，使用诸如Map-side join、Reduce-side join 或 Skew-Join等策略来优化。 -
限制Reducer输入：
为Reducer设置大小阈值，限制处理过多数据。 -
使用外部工具：
利用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));
}
}
