小编Mao*_*ron的帖子

我在AWS EMR上使用Scala运行Spark 2.2(Zeppling/spark-shell).

我正在尝试计算非常简单的计算:加载,过滤,缓存和计数大型数据集.我的数据包含4,500 GB(4.8 TB)ORC格式,包含51,317,951,565(510亿)行.

首先,我尝试使用以下群集进行处理:

1个主节点 - m4.xlarge - 4个cpu,16 gb内存

150个核心节点 - r3.xlarge - 4个cpu,29 gb内存

150个任务节点 - r3.xlarge - 4 cpu,29 gb Mem

但它失败了OutOfMemoryError.

当我查看Spark UI和Ganglia时,我看到在应用程序加载了超过80%的数据后,驱动程序节点变得太忙而执行程序停止工作(CPU使用率非常低),直到崩溃为止.

Ganglia主节点和工作节点的CPU使用率

然后我尝试执行相同的过程只是将驱动程序节点增加到:

1个主节点 - m4.2xlarge - 8个cpu,31 gb Mem

它成功了.

我不明白为什么驱动程序节点内存使用量在崩溃之前就已经完成了.AFAIK只有执行程序正在加载和处理任务,数据不应传递给主服务器.可能是什么原因呢？

1)Ganglia主节点用于第二个场景

2)Spark UI阶段

3)Spark UI DAG可视化

您可以在下面找到代码:

import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, SaveMode, SparkSession, DataFrame}
import org.apache.spark.sql.functions.{concat_ws, expr, lit, udf}
import org.apache.spark.storage.StorageLevel

val df = spark.sql("select * from default.level_1 where date_  >= ('2017-11-08') and date_ …

我使用的是 Airflow 1.9.0 版，我将拥有数百个 Dag。有没有办法将 Airflow UI 与文件夹、子文件夹一起排列，然后才将 Dag 放入其中？