仅在一个节点上运行的Dataproc Pyspark作业

Question

我的问题是我的pyspark作业没有并行运行。

代码和数据格式:
我的PySpark如下所示（显然是简化的）：

class TheThing:
    def __init__(self, dInputData, lDataInstance):
        # ...
    def does_the_thing(self):
        """About 0.01 seconds calculation time per row"""
        # ...
        return lProcessedData

#contains input data pre-processed from other RDDs
#done like this because one RDD cannot work with others inside its transformation
#is about 20-40MB in size
#everything in here loads and processes from BigQuery in about 7 minutes
dInputData = {'dPreloadedData': dPreloadedData}

#rddData contains about 3M rows
#is about 200MB large in csv format
#rddCalculated is about the same size as rddData
rddCalculated = (
    rddData
        .map(
            lambda l, dInputData=dInputData: TheThing(dInputData, l).does_the_thing()
        )
)

llCalculated = rddCalculated.collect()
#save as csv, export to storage

PySpark的全部要点是并行运行这个东西，显然不是这样。我在各种集群配置中运行了这些数据，最后一个配置是大量的，这时我注意到它是单一节点使用的。因此，为什么我的工作需要很长时间才能完成，而时间似乎与集群规模无关。

所有较小数据集的测试在我的本地机器和集群上都没有问题。我真的只是需要高档。

编辑
我更改了
llcalculated=rddcalculated.collect()
#...保存到csv并导出
到
rddcalculated.saveastextfile（“gs：//storage-bucket/results”）

而且只有一个节点还在做这项工作。

Answer 1

根据您是从GCS还是HDFS加载rdddata，默认拆分大小可能是64MB或128MB，这意味着您的200MB数据集只有2-4个分区。Spark这样做是因为典型的基本数据并行任务在数据中的搅动速度足够快，64MB-128MB意味着可能需要几十秒的处理时间，所以将并行度分成更小的块没有好处，因为启动开销将占主导地位。

在您的示例中，由于您与其他数据集连接，并且可能对每个记录执行相当重量级的计算，因此每MB的处理时间似乎要高得多。因此，您需要更多的分区，否则无论您有多少节点，Spark都不知道将其拆分为2-4个工作单元（如果每台机器都有多个核心，那么这些工作单元也可能被打包到一台机器上）。

因此，只需调用repartition:

rddCalculated = (
    rddData
        .repartition(200)
        .map(
            lambda l, dInputData=dInputData: TheThing(dInputData, l).does_the_thing()
        )
)

rddData = rddData.repartition(200)

或者，如果在读取时重新分区，您可能会有更好的效率：

rddData = sc.textFile("gs://storage-bucket/your-input-data", minPartitions=200)