问题描述

问题描述/异常栈

org.apache.spark.shuffle.FetchFailedException: failed to allocate 16777216 byte(s) of direct memory (used: 1073741824, max: 1073741824)

解决方案

增加 shuffle 分区数
比如 conf.spark.sql.shuffle.partitions=2001 (可根据实际数据量增加)

减少拉取数据并发度
比如 conf.spark.reducer.maxReqsInFlight=200

直接禁用 netty 的direct memory
conf.spark.shuffle.io.preferDirectBufs=false
conf.spark.network.io.preferDirectBufs=false

进阶方案

HighlyCompress的分区信息不准确，导致 CustomShuffleReader 把小分区合并成大分区的过程中发生了数据倾斜。
set spark.shuffle.minNumPartitionsToHighlyCompress=4096

注意：
conf.spark.sql.shuffle.partitions=4096 需要保持一致
spark.shuffle.io.preferDirectBufs=false
spark.network.io.preferDirectBufs=false
其余参数同上设置

问题原因

同时拉取大量的shuffle block，导致netty自己控制的内存超过限制

作者：焦巍