一、前言：OOM 的噩梦

本文将覆盖的 OOM 场景：

二、JVM 内存模型基础

2.1 JVM 内存结构

2.2 常见 OOM 类型

2.3 JVM 参数速查

# 堆内存设置
-Xms4g                    # 初始堆大小
-Xmx8g                    # 最大堆大小
-Xmn2g                    # 年轻代大小

# 元空间设置
-XX:MetaspaceSize=256m    # 初始元空间
-XX:MaxMetaspaceSize=512m # 最大元空间

# 直接内存
-XX:MaxDirectMemorySize=1g

# GC 相关
-XX:+UseG1GC              # 使用 G1 GC
-XX:MaxGCPauseMillis=200  # 目标暂停时间

# OOM 时 dump 堆
-XX:+HeapDumpOnOutOfMemoryError
-XX:HeapDumpPath=/tmp/heapdump.hprof

三、Spark OOM 全解析

3.1 Spark 内存模型

内存配置：

# Executor 内存
spark.executor.memory=8g
spark.executor.memoryOverhead=2g  # 堆外内存

# 内存比例
spark.memory.fraction=0.6         # 执行+存储 占比
spark.memory.storageFraction=0.5  # 存储在上面的占比

# 堆外内存
spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=2g

3.2 案例一：Shuffle OOM

问题代码：

// ❌ 大表 JOIN 大表，Shuffle 数据量爆炸
val result = tableA.join(tableB, "key")  // 两个表各10亿行

解决方案：

// ✅ 方案1：增加 Shuffle 分区数
spark.conf.set("spark.sql.shuffle.partitions", 2000)  // 默认200太小

// ✅ 方案2：增加 Executor 内存
spark.conf.set("spark.executor.memory", "16g")
spark.conf.set("spark.executor.memoryOverhead", "4g")

// ✅ 方案3：如果小表可以广播
val result = tableA.join(broadcast(smallTable), "key")

// ✅ 方案4：分批处理
val dates = (1 to 30).map(i => s"2024-01-$i")
dates.foreach { date =>
  val dailyResult = tableA.filter(col("dt") === date)
    .join(tableB.filter(col("dt") === date), "key")
  dailyResult.write.mode("append").save("/output")
}

3.3 案例二：数据倾斜导致 OOM

问题场景：

// ❌ 某个 user_id 有1000万条记录，其他用户只有几百条
df.groupBy("user_id").agg(sum("amount"))

解决方案：

// ✅ 方案1：加盐打散
val saltedDF = df.withColumn("salted_key", 
  concat(col("user_id"), lit("_"), (rand() * 10).cast("int")))

val step1 = saltedDF.groupBy("salted_key", "user_id")
  .agg(sum("amount").as("partial_sum"))

val result = step1.groupBy("user_id")
  .agg(sum("partial_sum").as("total_amount"))

// ✅ 方案2：单独处理热点 Key
val hotKeys = Set("hot_user_1", "hot_user_2")

val hotData = df.filter(col("user_id").isin(hotKeys.toSeq: _*))
val normalData = df.filter(!col("user_id").isin(hotKeys.toSeq: _*))

val hotResult = hotData.repartition(100, col("user_id"))
  .groupBy("user_id").agg(sum("amount"))

val normalResult = normalData.groupBy("user_id").agg(sum("amount"))

val result = hotResult.union(normalResult)

// ✅ 方案3：使用 AQE (Adaptive Query Execution)
spark.conf.set("spark.sql.adaptive.enabled", true)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", true)

3.4 案例三：广播变量 OOM

问题代码：

// ❌ 广播了一个超大的 Map
val bigMap = hugeDF.collect().map(r => r.getAs[String]("key") -> r).toMap
val broadcastMap = spark.sparkContext.broadcast(bigMap)  // Driver OOM!

解决方案：

// ✅ 方案1：限制广播大小
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 100 * 1024 * 1024)  // 100MB

// ✅ 方案2：使用 MapSide Join（适合维表）
// 将维表预先分发到每个分区
val dimensionDF = spark.read.parquet("/dimension_table")
val result = factDF.join(dimensionDF, Seq("dim_key"), "left")

// ✅ 方案3：如果必须用 Map，使用 mapPartitions
df.mapPartitions { iter =>
  // 每个分区独立加载需要的数据
  val localMap = loadLocalData()
  iter.map(row => processWithMap(row, localMap))
}

3.5 案例四：Driver OOM

问题代码：

// ❌ 收集大量数据到 Driver
val allData = hugeDF.collect()  // Driver OOM!

// ❌ 保存过程中产生太多小文件的元数据
df.repartition(100000).write.save("/output")  // 10万个文件的元数据

解决方案：

// ✅ 方案1：增加 Driver 内存
spark.conf.set("spark.driver.memory", "8g")
spark.conf.set("spark.driver.maxResultSize", "4g")

// ✅ 方案2：避免 collect，使用 take 或 limit
val sample = hugeDF.take(1000)  // 只取部分

// ✅ 方案3：直接写入存储，不要收集
hugeDF.write.parquet("/output")  // 不经过 Driver

// ✅ 方案4：合理控制输出文件数
df.repartition(200).write.save("/output")  // 控制文件数量

3.6 Spark OOM 排查流程

四、Flink OOM 全解析

4.1 Flink 内存模型

内存配置：

# flink-conf.yaml
taskmanager.memory.process.size: 8g
taskmanager.memory.framework.heap.size: 128m
taskmanager.memory.task.heap.size: 4g
taskmanager.memory.managed.size: 2g
taskmanager.memory.network.fraction: 0.1
taskmanager.memory.jvm-overhead.fraction: 0.1

4.2 案例一：State 过大 OOM

问题代码：

// ❌ 在 State 中存储了整个列表，没有清理
public class BadStatefulFunction extends KeyedProcessFunction<String, Event, Result> {
    private ListState<Event> historyState;
    
    @Override
    public void processElement(Event event, Context ctx, Collector<Result> out) {
        historyState.add(event);  // 无限增长！
        // 没有清理逻辑...
    }
}

解决方案：

// ✅ 方案1：设置 State TTL
StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(7))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter(1000)  // RocksDB 后台清理
    .build();

ListStateDescriptor<Event> descriptor = new ListStateDescriptor<>("history", Event.class);
descriptor.enableTimeToLive(ttlConfig);

// ✅ 方案2：使用 RocksDB State Backend（状态可以溢出到磁盘）
env.setStateBackend(new EmbeddedRocksDBStateBackend());

// ✅ 方案3：定期清理 State
@Override
public void processElement(Event event, Context ctx, Collector<Result> out) {
    historyState.add(event);
    
    // 只保留最近 1000 条
    List<Event> history = Lists.newArrayList(historyState.get());
    if (history.size() > 1000) {
        historyState.clear();
        historyState.addAll(history.subList(history.size() - 1000, history.size()));
    }
}

4.3 案例二：窗口积压 OOM

问题场景：

// ❌ 会话窗口 gap 设置过大，窗口迟迟不关闭
dataStream
    .keyBy(Event::getUserId)
    .window(EventTimeSessionWindows.withGap(Time.hours(24)))  // 24小时不活动才关闭
    .aggregate(new MyAggregator());  // 窗口内数据持续增长

解决方案：

// ✅ 方案1：限制窗口大小
dataStream
    .keyBy(Event::getUserId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))  // 缩短 gap
    .trigger(CountTrigger.of(10000))  // 或者达到数量就触发
    .aggregate(new MyAggregator());

// ✅ 方案2：使用增量聚合，不存储原始数据
dataStream
    .keyBy(Event::getUserId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .aggregate(
        new AggregateFunction<Event, Accumulator, Result>() {
            // 只存储聚合结果，不存储原始数据
        }
    );

// ✅ 方案3：设置允许延迟并清理
dataStream
    .keyBy(Event::getUserId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .allowedLateness(Time.minutes(10))  // 允许延迟
    .sideOutputLateData(lateOutputTag)  // 延迟数据旁路输出
    .aggregate(new MyAggregator());

4.4 案例三：Checkpoint 超时导致内存积压

解决方案：

// ✅ 方案1：使用 Unaligned Checkpoint (Flink 1.11+)
env.getCheckpointConfig().enableUnalignedCheckpoints();

// ✅ 方案2：调整 Checkpoint 参数
env.enableCheckpointing(60000);  // 60秒间隔
env.getCheckpointConfig().setCheckpointTimeout(300000);  // 5分钟超时
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);  // 最小间隔

// ✅ 方案3：增加 Network Buffer
// flink-conf.yaml
taskmanager.memory.network.min: 256mb
taskmanager.memory.network.max: 1gb

4.5 Flink OOM 排查流程

五、Hive/Presto OOM

5.1 Hive OOM 场景

5.2 Hive 内存配置

-- Map 端内存
SET mapreduce.map.memory.mb=4096;
SET mapreduce.map.java.opts=-Xmx3276m;

-- Reduce 端内存
SET mapreduce.reduce.memory.mb=8192;
SET mapreduce.reduce.java.opts=-Xmx6553m;

-- 容器内存
SET yarn.nodemanager.resource.memory-mb=65536;

-- 处理数据倾斜
SET hive.optimize.skewjoin=true;
SET hive.skewjoin.key=100000;

-- Map Join 优化
SET hive.auto.convert.join=true;
SET hive.mapjoin.smalltable.filesize=50000000;

5.3 Presto OOM 处理

-- 查询内存限制
SET query_max_memory = '50GB';
SET query_max_memory_per_node = '10GB';

-- 聚合内存
SET aggregation_operator_unspill_memory_limit = '4GB';

-- 分页处理
SELECT * FROM huge_table 
WHERE dt = '2024-01-15'
LIMIT 1000000 OFFSET 0;  -- 分批查询

六、通用排查方法

6.1 获取内存 Dump

# JVM 堆 Dump
jmap -dump:format=b,file=heap.hprof <pid>

# 在 OOM 时自动 Dump（启动参数）
-XX:+HeapDumpOnOutOfMemoryError
-XX:HeapDumpPath=/tmp/heapdump.hprof

# Spark 配置
spark.executor.extraJavaOptions=-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/

6.2 分析工具

6.3 Arthas 常用命令

# 安装
curl -O https://arthas.aliyun.com/arthas-boot.jar
java -jar arthas-boot.jar

# 常用命令
dashboard          # 总览
thread -n 5        # 最忙的5个线程
heapdump /tmp/dump.hprof  # 堆 Dump
memory             # 内存信息
jvm                # JVM 信息

七、OOM 预防清单

7.1 Spark 预防清单

7.2 Flink 预防清单

7.3 通用预防措施

八、OOM 案例速查表

8.1 错误信息速查

8.2 快速定位公式

OOM 定位 = 确定组件 + 确定阶段 + 分析堆栈 + 重现问题

九、总结

9.1 核心要点

9.2 一句话总结

预防：合理配置 + 代码优化 + 监控告警

定位：错误类型 + 组件阶段 + 堆栈分析

解决：增加内存是下策，优化代码是上策

9.3 记住这个口诀

Spark OOM 三板斧：增分区、加内存、治倾斜

Flink OOM 三板斧：限状态、用RocksDB、调Checkpoint

通用 OOM 三板斧：看日志、dump堆、找泄漏

十、结语

恭喜你完成了「从入门到放弃」大数据系列的学习！

这个系列我们覆盖了：

希望这些内容能帮助你在大数据领域少踩坑、快成长！