在Flink on YARN集群中提交第二个Job时出现Task的Java Heap Size OOM（内存溢出），核心原因是集群资源不足（第一个Job已占用大量资源，第二个Job分配的内存不够）或Job资源配置不合理（单个Job申请资源过多，或内存参数设置与实际需求不匹配）。解决需从“资源排查”“Job配置优化”“集群资源扩容”三个维度入手，具体步骤如下：

一、先排查核心原因：确认资源占用与配置是否匹配

1. 检查YARN集群当前资源使用情况

OOM的直接触发点是“第二个Job的TaskManager无法获得足够内存”，需先确认YARN集群是否有足够剩余资源：

• 通过YARN UI（通常是http://<yarn-resourcemanager>:8088）查看：
  • 总资源（Total Memory/Total vCores）、已使用资源（Used Memory/Used vCores）、剩余资源（Available Memory/Available vCores）。
  • 若剩余内存远小于第二个Job申请的内存（比如第二个Job申请10GB，剩余仅5GB），则是集群资源不足。
  • 若剩余资源充足，但第二个Job仍OOM，可能是Job自身内存配置不合理（如单个TaskManager堆内存设置过小，或并行度过高导致单Task内存压力大）。

2. 检查已运行Job和新Job的资源配置

Flink Job在YARN上运行时，资源申请通过以下参数控制（提交Job时指定），需确认是否存在“单个Job申请资源过多”或“内存分配不合理”：

• yarn.containers：申请的YARN Container数量（对应Flink的TaskManager数量）。
• taskmanager.memory.process.size：单个TaskManager的总进程内存（包括JVM堆、堆外、网络等所有内存，推荐用此参数统一配置）。
• taskmanager.numberOfTaskSlots：单个TaskManager的Slot数量（决定并行度上限）。
• parallelism.default：Job默认并行度（并行度越高，单个Task的内存压力可能越大，若总内存固定，并行度过高会导致单Task内存不足）。

例如：若第一个Job已占用80%的集群内存，第二个Job仍按“默认大内存”申请，会导致分配到的TaskManager内存不足，触发OOM。

二、核心解决方法：优化Job资源配置（优先操作）

若集群总资源有限（无法立即扩容），需通过调整Job的资源配置，减少单个Job的资源占用，或让内存分配更合理。

1. 降低单个Job的资源申请量

• 减少并行度：并行度（parallelism.default）越高，需要的Task数量越多，每个Task的内存分配可能被稀释。若Job逻辑不依赖高并行度（如非实时高吞吐场景），可降低并行度：

  # 提交Job时指定并行度为原来的1/2（例如从10降为5）
  flink run -m yarn-cluster -p 5 ...
  

• 降低单个TaskManager的内存：若TaskManager内存配置过大（如taskmanager.memory.process.size=16g），但实际Job不需要这么多，可减小：

  # 单个TaskManager总内存从16g降为8g（根据Job实际需求调整）
  flink run -m yarn-cluster -ytm 8g ...
  

  （-ytm 是 taskmanager.memory.process.size 的简写，推荐用此参数控制总内存，Flink会自动分配堆内/堆外内存比例）

• 减少TaskManager数量：若Job并行度不高，可减少yarn.containers（TaskManager数量），避免资源碎片化：

  # 申请3个TaskManager（默认可能更多）
  flink run -m yarn-cluster -yn 3 ...
  

2. 优化TaskManager的堆内存分配

Flink的TaskManager内存中，Task的Java堆内存（Task Heap） 是运行用户代码的核心内存（OOM通常指此部分不足）。若总内存固定，可通过参数调整堆内存占比：

• 若使用 taskmanager.memory.process.size 统一配置总内存，Flink会自动分配堆内存（默认约占总内存的40%-60%），可通过以下参数手动调整堆内存下限：

  # 单个Task的堆内存至少2g（避免被其他内存挤占）
  flink run -m yarn-cluster -D taskmanager.memory.task.heap.min=2g ...
  

• 若明确知道Task需要更多堆内存，可直接指定Task堆内存（需注意总内存是否足够）：

  # 单个Task的堆内存设置为3g（总内存需大于此值，避免超配）
  flink run -m yarn-cluster -D taskmanager.memory.task.heap.size=3g ...
  

3. 避免两个Job“资源争抢”：错开资源高峰或隔离队列

• YARN队列隔离：若YARN配置了队列（Queue），可将两个Job提交到不同队列，通过队列资源限制避免互相抢占。例如：

  • 为第一个Job分配“高资源队列”（如queueA，内存占比60%），第二个Job提交到“低资源队列”（queueB，内存占比40%）。
  • 提交时指定队列：flink run -m yarn-cluster -yq queueB ...

• 错峰运行：若两个Job非实时必须同时运行，可调整调度时间（如一个白天运行，一个夜间运行），避免资源冲突。

三、集群层面优化：提升资源可用量

若Job无法通过降低配置满足需求（如业务要求高并行度、大内存），需从集群层面扩容或调整资源配置。

1. 增加YARN集群的总资源

• 向YARN集群添加新的NodeManager节点（物理机/虚拟机），提升总内存和CPU容量。
• 若现有节点有未分配的资源（如NodeManager配置的yarn.nodemanager.resource.memory-mb小于节点实际内存），可修改yarn-site.xml调整：

  <!-- 例如节点实际内存64GB，可将配置改为60GB（保留部分给系统） -->
  <property>
    <name>yarn.nodemanager.resource.memory-mb</name>
    <value>61440</value> <!-- 60GB=60*1024MB -->
  </property>
  

  修改后重启NodeManager生效。

2. 调整YARN资源分配策略

• 降低YARN的最小内存分配单位（yarn.scheduler.minimum-allocation-mb）：若默认最小分配为2GB，而Job需要1.5GB，可调整为1GB，避免资源浪费：

  <property>
    <name>yarn.scheduler.minimum-allocation-mb</name>
    <value>1024</value> <!-- 1GB -->
  </property>
  

• 关闭YARN的资源严格限制（谨慎使用）：若YARN启用了yarn.nodemanager.pmem-check-enabled（物理内存检查），可临时关闭（避免Task使用稍超申请的内存被Kill），但需确保节点总内存充足：

  <property>
    <name>yarn.nodemanager.pmem-check-enabled</name>
    <value>false</value>
  </property>
  

四、排查是否存在内存泄漏（非资源问题）

若以上方法无效，需确认是否是Job自身代码存在内存泄漏（如未释放的大对象、无限增长的集合），导致即使内存充足也会OOM：

• 通过Flink UI（http://<jobmanager>:8081）查看Task的“Memory Usage”监控，若内存持续增长且不释放，可能是泄漏。
• 开启Flink的内存采样（-D taskmanager.debug.memory=true），或在OOM时生成Heap Dump（-XX:+HeapDumpOnOutOfMemoryError），分析泄漏对象。

总结操作步骤

1. 先通过YARN UI确认集群剩余资源，判断是“资源不足”还是“配置不合理”；
2. 若资源不足：降低现有Job的并行度/内存配置，或隔离队列；
3. 若配置不合理：调整TaskManager堆内存参数（如taskmanager.memory.task.heap.size）；
4. 长期解决：扩容YARN集群资源，或优化Job代码避免内存浪费。

通过以上步骤，可逐步定位并解决“第二个Job提交时Task OOM”的问题。核心原则是“让资源分配与实际需求匹配”，避免过度占用或配置不足。