我正在构建一个相对简单的Spark应用程序，其逻辑大致如下：

val file1 = sc.textFile("s3://file1/*")
val file2 = sc.textFile("s3://file2/*")
// map over files
val file1Map = file1.map(word => (word, "val1"))
val file2Map = file2.map(differentword => (differentword, "val2"))
val unionRdd = file1Map.union(file2Map)
val groupedUnion = unionRdd.groupByKey()
val output = groupedUnion.map(tuple => {
    // do something that requires all the values, return new object
    if(oneThingIsTrue) tuple._1 else "null"
}).filter(line => line != "null")
output.saveAsTextFile("s3://newfile/")

这个问题与我在运行较大的数据集时不工作有关。当数据集大约为700GB时，我可以运行它而不出错。当我将它加倍到1.6TB时，作业将在超时前完成一半。下面是错误日志：

INFO MapOutputTrackerWorker: Don't have map outputs for shuffle 0, fetching them
INFO MapOutputTrackerWorker: Doing the fetch; tracker endpoint = NettyRpcEndpointRef(spark://MapOutputTracker@172.31.4.36:39743)
ERROR MapOutputTrackerWorker: Error communicating with MapOutputTracker
    org.apache.spark.rpc.RpcTimeoutException: Futures timed out after [800 seconds]. This timeout is controlled by spark.network.timeout

我试过将网络超时时间增加到800秒和1600秒，但这只是延长了错误时间。我在10r4.2xl上运行代码，每个内核有8个内核，内存为62GB。我将EBS设置为3TB存储空间。我在Amazon EMR中通过Zeppelin运行此代码。
有人能帮我调试一下吗？集群的CPU使用率一直接近90%，直到它达到一半，然后完全降回0。另一个有趣的事情是，它看起来像是在第二阶段洗牌时失败了。正如您从跟踪中看到的，它正在执行获取操作，但从未获得它。
这是一张来自Ganglia的照片。