1. 数据倾斜：Hadoop开发中的"头号杀手"

数据倾斜是每个Hadoop开发者都会遇到的典型问题。简单来说，就是某些节点处理的数据量远远超过其他节点，导致整个集群像瘸腿的运动员一样无法高效运转。我在实际项目中见过最夸张的情况：一个Reduce任务处理了90%的数据，而其他节点早早完成任务却在"围观"。

常见的数据倾斜场景：

• 电商平台统计商品点击量时，爆款商品的数据量可能是普通商品的数万倍
• 社交网络分析中，明星用户的关注关系数据远超普通用户
• 日志分析时，某些异常状态的日志条目突然激增

数据倾斜会导致三大问题：

1. 个别节点长时间运行无法结束
2. 其他节点资源闲置造成浪费
3. 整体作业时间被最长任务拖累

1.1 诊断数据倾斜的实用技巧

发现数据倾斜不能只靠猜，这里分享几个我常用的诊断方法：

# 查看任务计数器，重点关注不同reduce处理的数据量差异
hadoop job -counter <job_id> 'org.apache.hadoop.mapred.Task$Counter' REDUCE_INPUT_RECORDS

# 使用Hadoop Web UI观察各个reduce任务的进度差异
http://<jobtracker-host>:50030/jobdetails.jsp?jobid=<job_id>

更直观的方法是在Mapper端添加日志：

// 在map方法中添加计数器
context.getCounter("SKEW_DETECT", key.toString()).increment(1);

运行后通过hadoop job -counter查看各key的分布情况。

1.2 六种数据倾斜解决方案实战

根据不同的业务场景，我总结出六种应对策略：

方案1：Combiner前置聚合

// 自定义Combiner实现局部聚合
public static class MyCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
    public void reduce(Text key, Iterable<IntWritable> values, Context context) {
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        context.write(key, new IntWritable(sum));
    }
}

在job配置中添加：

job.setCombinerClass(MyCombiner.class);

方案2：随机前缀再聚合 适用于group by场景，分两阶段处理：

-- 第一阶段：给key添加随机前缀
SELECT concat(key, '_', rand()) as new_key, value 
FROM table 
GROUP BY concat(key, '_', rand()), value

-- 第二阶段：去除前缀全局聚合
SELECT substring(key, 1, position('_' in key)-1) as original_key, sum(value)
FROM stage1_result
GROUP BY substring(key, 1, position('_' in key)-1)

方案3：Map Join替代Reduce Join 当小表可以装入内存时：

// 在setup阶段加载小表到内存
Map<String, String> smallTable = new HashMap<>();
try (BufferedReader br = new BufferedReader(new FileReader("small_table.txt"))) {
    String line;
    while ((line = br.readLine()) != null) {
        String[] parts = line.split("\t");
        smallTable.put(parts[0], parts[1]);
    }
}

// 在map方法中直接关联
protected void map(LongWritable key, Text value, Context context) {
    String[] fields = value.toString().split("\t");
    String joinedValue = smallTable.get(fields[0]);
    // ...输出结果
}

其他方案还包括： 4. 自定义分区算法避免热点 5. 调整reduce任务数分散压力 6. 对倾斜key单独处理

提示：实际应用中常需要组合多种方案，比如先用随机前缀分散数据，再配合Combiner减少网络传输

2. MapReduce性能优化全攻略

2.1 资源配置的黄金法则

资源配置不当是性能差的常见原因。根据集群规模和数据量，我总结出这些经验值：

参数名	小集群(10节点)	中集群(50节点)	大集群(100+节点)
mapreduce.map.memory.mb	2G	4G	8G
mapreduce.reduce.memory.mb	4G	8G	16G
mapreduce.task.io.sort.mb	512M	1G	2G
mapreduce.map.sort.spill.percent	0.8	0.8	0.8

配置示例：

<!-- mapred-site.xml -->
<property>
    <name>mapreduce.map.memory.mb</name>
    <value>4096</value>
</property>
<property>
    <name>mapreduce.reduce.memory.mb</name>
    <value>8192</value>
</property>

2.2 切片与并行度的艺术

切片大小直接影响Map任务数，计算公式：

split_size = max(min_size, min(max_size, block_size))

优化案例：

// 处理大量小文件时，使用CombineTextInputFormat
job.setInputFormatClass(CombineTextInputFormat.class);
CombineTextInputFormat.setMaxInputSplitSize(job, 128 * 1024 * 1024); // 128MB
CombineTextInputFormat.setMinInputSplitSize(job, 64 * 1024 * 1024); // 64MB

Reduce任务数设置经验：

• 建议为集群可用reduce slot的0.95~1.75倍
• 每个reduce处理数据量建议在1~5GB之间

// 根据数据量动态设置reduce数
long dataSize = job.getConfiguration().getLong("mapreduce.input.fileinputformat.split.size", 128 * 1024 * 1024);
int numReduces = (int) (totalInputSize / (5 * 1024 * 1024 * 1024L)) + 1;
job.setNumReduceTasks(Math.min(numReduces, 100));

2.3 Shuffle阶段调优实战

Shuffle是MR中最耗时的阶段，关键优化点：

1. 压缩中间数据：

<property>
    <name>mapreduce.map.output.compress</name>
    <value>true</value>
</property>
<property>
    <name>mapreduce.map.output.compress.codec</name>
    <value>org.apache.hadoop.io.compress.SnappyCodec</value>
</property>

2. 调整缓冲区参数：

<property>
    <name>mapreduce.task.io.sort.mb</name>
    <value>1024</value>  <!-- 排序缓冲区大小 -->
</property>
<property>
    <name>mapreduce.map.sort.spill.percent</name>
    <value>0.90</value>  <!-- 溢出阈值 -->
</property>

3. 合并小文件：

<property>
    <name>mapreduce.task.io.sort.factor</name>
    <value>100</value>  <!-- 一次合并的文件数 -->
</property>

3. HDFS存储优化策略

3.1 小文件合并最佳实践

HDFS小文件问题就像把图书馆的书全部拆成单页存放，解决方法：

方案1：Har归档

hadoop archive -archiveName data.har -p /input /output

方案2：MapReduce合并

public class SmallFilesToSequenceFile extends Configured implements Tool {
    public int run(String[] args) throws Exception {
        Job job = Job.getInstance(getConf());
        job.setInputFormatClass(WholeFileInputFormat.class); // 自定义InputFormat
        // ...其他配置
        return job.waitForCompletion(true) ? 0 : 1;
    }
}

方案3：Hive外部表合并

CREATE EXTERNAL TABLE consolidated_table 
STORED AS ORC
LOCATION '/path/to/consolidated'
AS SELECT * FROM source_table;

3.2 存储策略选择

HDFS支持多种存储策略，根据数据热度配置：

存储策略	适用场景	配置方法
HOT	频繁访问的热数据	默认策略
WARM	偶尔访问的温数据	hdfs storagepolicies -setStoragePolicy -path /path -policy WARM
COLD	极少访问的冷数据	同上，policy改为COLD
ALL_SSD	超高频访问的关键数据	同上，policy改为ALL_SSD

查看文件存储策略：

hdfs storagepolicies -getStoragePolicy -path /user/data/file1

4. YARN资源调度实战

4.1 调度器选型指南

YARN三种调度器对比：

特性	FIFO	Capacity Scheduler	Fair Scheduler
资源分配方式	先进先出	队列容量保证	公平共享
资源利用率	低	中	高
适合场景	测试环境	多租户生产环境	灵活多变负载
关键配置	无	yarn.scheduler.capacity.root.queues	yarn.scheduler.fair.allocation.file

Fair Scheduler配置示例(fair-scheduler.xml)：

<allocations>
  <queue name="etl">
    <minResources>10000 mb,10vcores</minResources>
    <maxResources>90000 mb,100vcores</maxResources>
    <schedulingPolicy>fair</schedulingPolicy>
  </queue>
  <queue name="ad_hoc">
    <minResources>20000 mb,20vcores</minResources>
    <maxResources>100000 mb,200vcores</maxResources>
  </queue>
</allocations>

4.2 资源隔离与限制

防止异常任务耗尽资源的关键配置：

<!-- 限制单个容器最大资源 -->
<property>
    <name>yarn.scheduler.maximum-allocation-mb</name>
    <value>16384</value>  <!-- 16GB -->
</property>
<property>
    <name>yarn.scheduler.maximum-allocation-vcores</name>
    <value>8</value>
</property>

<!-- 开启资源监控 -->
<property>
    <name>yarn.nodemanager.pmem-check-enabled</name>
    <value>true</value>
</property>
<property>
    <name>yarn.nodemanager.vmem-check-enabled</name>
    <value>true</value>
</property>

5. 企业级调优案例解析

某电商平台双十一大促期间，订单分析作业出现严重数据倾斜。通过以下步骤解决：

1. 问题诊断：

   • 发现99%的订单集中在10%的商品
   • 部分reduce任务处理时间是其他的100倍

2. 解决方案：

   // 自定义分区器分散热点商品
   public class SkewAwarePartitioner extends Partitioner<Text, IntWritable> {
       private Random rand = new Random();
       
       @Override
       public int getPartition(Text key, IntWritable value, int numPartitions) {
           if(isHotItem(key.toString())) {  // 判断是否热点商品
               return rand.nextInt(numPartitions);  // 随机分配
           }
           return Math.abs(key.hashCode()) % numPartitions;  // 常规哈希
       }
   }
   

3. 效果对比：

   • 优化前：2小时15分钟完成
   • 优化后：38分钟完成
   • 资源消耗减少60%

另一个日志分析案例中，通过以下组合优化提升性能：

• 使用Snappy压缩map输出
• 调整io.sort.mb从256MB到1GB
• 设置mapreduce.job.jvm.numtasks=10实现JVM重用
• 采用CombineTextInputFormat合并小日志文件

最终效果：

• 作业运行时间从4.2小时降至1.5小时
• Shuffle数据传输量减少70%
• 集群整体负载更加均衡