如何并行处理多个pyspark数据帧

Question

我有一个包含数百万条记录和数百列的 pyspark 数据框（以下示例）

clm1, clm2, clm3
code1,xyz,123
code2,abc,345
code1,qwe,456

我想根据 clm1 将其划分为多个数据帧，即 clm1=code1 的单独数据帧和 clm1=code2 的单独数据帧等等，然后处理它们并将结果写入单独的文件中。我想并行执行此操作以加快该过程。我正在使用以下代码：

S1 = myclass("code1")
S2 = myclass("code2")


t1 = multiprocessing.Process(target=S1.processdata,args=(df,))
t2 = multiprocessing.Process(target=S2.processdata,args=(df,))
t1.start()
t2.start()

t1.join()
t2.join()

但我遇到了错误

Method __getstate__([]) does not exist

如果我使用 threading.Thread 而不是 multiprocessing.Process 它工作正常，但这似乎并没有减少总时间

Answer 1

关于错误

方法getstate([])不存在

这是一个py4j.Py4JException。 multiprocessing.Process 出现此错误，因为此模块使用进程。另一方面 threading.Thread 使用使用相同内存的线程，因此它们可以共享数据帧对象。

也看看那个 SO 问题答案：Multiprocessing vs Threading Python

---

一般建议

我知道您可能是 Spark 世界的新手，我建议您使用我的解决方案来解决您的问题。您询问了如何进行多处理，但如果您有 Spark，这可能不是最佳实践。

您拥有 Spark-并行处理框架，您无需手动并行化您的任务。

Spark 设计用于集群中的并行计算，但它在大型单节点中运行得非常好。多处理库在 Python 计算任务中很有用，在 Spark/Pyspark 中，所有计算在 JVM 中并行运行。

在 python_code.py 中

import pyspark.sql.functions as f


# JOB 1
df1 = df.filter(f.col('clm1')=='code1')
... many transformations
df1.write.format('..')..

# JOB 2
df2 = df.filter(f.col('clm1')=='code2')
... many transformations
df2.write.format('..')..

然后使用你的所有核心（* = 所有核心）通过 spark-submit 运行此代码

# Run application locally on all cores
./bin/spark-submit --master local[*] python_code.py

通过这种方法，您可以使用 Spark 的强大功能。这些作业将按顺序执行，但您将拥有： 始终使用 CPU 并行处理 计算时间较短