Spark 本地模式 - 所有作业仅使用一个 CPU 内核

Question

我们在单个 AWS EC2 实例上以本地模式运行 Spark Java，使用

"local[*]"

但是，使用 New Relic 工具和简单的“顶部”进行分析表明，我们的 16 核机器中只有一个 CPU 内核用于我们编写的三个不同的 Java spark 作业（我们还尝试了不同的 AWS 实例但只使用了一个核心）。

Runtime.getRuntime().availableProcessors() 报告 16 个处理器和 sparkContext.defaultParallelism() 也报告了 16 个。

我查看了各种 Stackoverflow 本地模式问题，但似乎都没有解决问题。

非常感谢任何建议。

谢谢

编辑：过程

1) 使用 sqlContext 将使用 com.databricks.spark.csv 的压缩 CSV 文件 1 从磁盘 (S3) 读取到 DataFrame DF1。

2) 使用 sqlContext 将使用 com.databricks.spark.csv 的压缩 CSV 文件 2 从磁盘 (S3) 读取到 DataFrame DF2。

3) 使用 DF1.toJavaRDD().mapToPair(返回元组的新映射函数>) RDD1

4) 使用 DF2.toJavaRDD().mapToPair(返回元组的新映射函数>) RDD2

5) 在 RDD 上调用 union

6) 在联合的 RDD 上调用 reduceByKey() 以“按键合并”因此有一个元组>) 只有一个特定键的实例（因为相同的键出现在 RDD1 和 RDD2 中）。

7) 调用 .values().map(new mapping 函数，它遍历提供的 List 中的所有项目并根据需要合并它们以返回相同或更小的长度的 List

8) 调用 .flatMap() 获取 RDD

9) 使用 sqlContext 从 DomainClass 类型的平面地图创建 DataFrame

10) 使用 DF.coalease(1).write() 将 DF 作为 gzipped CSV 写入 S3。

Answer 1

我认为您的问题是您的 CSV 文件已压缩。当 Spark 读取文件时，它会并行加载它们，但它只能在文件编解码器是可拆分的情况下执行此操作*。纯（非压缩）文本和镶木地板是可拆分的，以及基因组学（我的领域）中使用的bgzip 编解码器。您的整个文件最终都在一个分区中。

尝试解压缩 csv.gz 文件并再次运行。我想你会看到更好的结果！

• 可拆分格式意味着，如果给定一个任意文件偏移量来开始读取，则可以在块中找到下一条记录的开头并对其进行解释。 Gzip 文件不可拆分。

编辑：我在我的机器上复制了这种行为。在 3G 压缩文本文件上使用 sc.textFile 会产生 1 个分区。