我在 Linux 机器上的一个目录中有大约 200 个文件，名为 part-0001、part-0002 等等。每个都有大约一百万行和相同的列(称它们为“a”、“b”等)。让对 'a','b' 成为每一行的键(有很多重复项)。
同时，我搭建了一个Spark 2.2.0集群，一主两从，共42核可用。地址是 spark://XXX.YYY.com:7077。
然后我使用 PySpark 连接到集群并计算每个唯一对的 200 个文件的计数，如下所示。
from pyspark import SparkContext
from pyspark.sql import SQLContext
import pandas as pd
sc = SparkContext("spark://XXX.YYY.com:7077")
sqlContext = SQLContext(sc)
data_path = "/location/to/my/data/part-*"
sparkdf = sqlContext.read.csv(path=data_path, header=True)
dfgrouped = sparkdf.groupBy(['a','b'])
counts_by_group = dfgrouped.count()
这是可行的，因为我可以看到 Spark 通过一系列消息进行处理，并且它确实返回了看起来合理的结果。
问题:在执行此计算时，top 没有显示任何证据表明从内核正在做任何事情。似乎没有任何并行化。每个从站都有一个在作业之前存在的相关 Java 进程(加上来自其他用户的进程和后台系统进程)。所以看起来主人正在做所有的工作。考虑到有 200 个奇怪的文件，我曾预计在每台从机上运行 21 个进程，直到事情结束(这 我在显式调用 parallelize 时看到的是遵循 count = sc.parallelize(c=range(1, niters + 1), numSlices=ncores).map(f).reduce(add) 在单独的实现中。
问题:如何确保 Spark 实际上并行化计数？我希望每个核心获取一个或多个文件，对它在文件中看到的对进行计数，然后将单个结果缩减为单个 DataFrame。我不应该在顶部看到这个吗？我是否需要显式调用并行化？
(FWIW，我看过使用分区的示例，但我的理解是这用于在单个文件的 block 上分配处理。我的情况是我有很多文件。)
提前致谢。
请您参考如下方法:
TL;DR您的部署可能没有问题。
I had expected to see 21 processes running
除非您专门将 Spark 配置为每个执行程序 JVM 使用一个内核，否则没有理由发生这种情况。与您在问题中提到的 RDD 示例不同，DataFrame API 根本不使用 Python worker，Python UserDefinedFunctions 除外。
同时，JVM 执行器使用线程而不是成熟的系统进程(PySpark 使用后者来避免 GIL )。此外，独立模式下的默认 spark.executor.cores 等于 the available cores on the worker 的数量.因此，如果没有额外的配置，您应该会看到两个执行程序 JVM，每个使用 21 个数据处理线程。
总的来说，您应该检查 Spark UI，如果您看到分配给执行程序的任务，一切都应该没问题。
