Python / PySpark并行处理示例

Posted 2021-05-03

我无法理解如何在我的python脚本中利用并行处理的强大功能。

我有十亿行食品数据库，例如：

date,item,category,number_sold
2017-01-01,bagel,bread,10
2017-03-03,skittles,candy,5

有100个类别。 （面包，糖果等）我的剧本：

1. Iterates list of 100 categories
2. Filter pyspark dataframe for the subset (e.g. category == 'bread')
3. Run aggregate calculations on subset
4. Generate 1 row of stats and appends to summary file

我需要在迭代循环中添加什么来触发多处理？ pyspark会自动执行此操作吗？当我仅使用Pandas运行时，脚本在等待查询每个类别子集时被阻止（没有做任何事情）。理想情况下，该流程应过滤一个类别的数据框，并同时运行另一个类别的计算。

提前致谢！

答案

第一个短列表项（步骤2）并在单独的线程中处理它们。在python中实现多线程非常容易。将每个报告写在一个单独的文件中，最后填写所有文件并进行最终报告。