Spark - 我无法在本地模式下增加任务数量

Question

我尝试提交我的申请并通过不同的组合更改我的代码中的coalese[k]：

首先，我从本地磁盘读取了一些数据：

val df = spark.read.option("encoding", "gbk").option("wholeFile",true).option("multiline",true).option("sep", "|+|").schema(schema).csv("file:///path/to/foo.txt")

情况1

我认为 local[*] 意味着总共有 56 个内核。我指定 4 * 4 = 16 个任务：

火花提交：

spark-submit --master local[*] --class foo --driver-memory-8g --executor-memory 4g --executor-cores 4 --num-executors 4 foo.jar

spark.write：

df.coalesce(16).write.mode("overwrite").partitionBy("date").orc("hdfs://xxx:9000/user/hive/warehouse/ods/foo")

但是当我查看 spark 历史日志服务器 UI 时，只有 1 个任务。在数据集中，“日期”列只有一个值。

所以我尝试了另一种组合并删除了partitionBy：

情况2

火花提交：

spark-submit --master local[*] --class foo foo.jar

spark.write：

df.coalesce(16).write.mode("overwrite").orc("hdfs://xxxx:9000/user/hive/warehouse/ods/foo")

但历史服务器显示仍然只有 1 个任务。

我的本​​地机器上有 56 个内核和 256GB 内存。 我知道在本地模式下，spark 会为驱动程序和执行程序创建一个 JVM，因此这意味着我们有一个执行程序，其内核数（假设为 56）我们的计算机（如果我们使用 Local[*] 运行它）。

以下是问题：

1. 谁能解释一下为什么我的任务编号总是 1？
2. 如何增加任务数量以便利用并行性？
3. 我的本地文件会被读入不同的分区吗？

Answer 1

Spark 只能通过一个执行器读取 csv 文件，因为只有一个文件。

与位于分布式文件系统（如 HDFS）中的文件相比，单个文件可以存储在多个分区中。这意味着您生成的 Dataframe df 只有一个分区。您可以使用df.rdd.getNumPartitions 进行检查。另请参阅我对How is a Spark Dataframe partitioned by default?的回答

请注意，coalesce 将折叠同一工作人员上的分区，因此调用 coalesce(16) 根本不会产生任何影响，因为您的 Dataframe 的一个分区已经位于单个工作人员上。

为了提高并行度，您可能需要改用repartition(16)。