如何使用 PySpark 并行化我的文件处理程序

Question

我现在有一个大型 python 项目，其中驱动程序有一个函数，该函数使用 for 循环遍历我的 GCP（谷歌云平台）存储桶上的每个文件。我正在使用 CLI 将作业提交到 GCP 并让作业在 GCP 上运行。

对于在此 for 循环中遍历的每个文件，我正在调用一个函数 parse_file(...) 来解析文件并调用处理此文件的其他函数的序列。

整个项目运行需要几分钟，速度很慢，而且驱动程序还没有使用太多的 PySpark。问题是该文件级 for 循环中的每个 parse_file(...) 都按顺序执行。是否可以使用 PySpark 并行化该文件级 for 循环以对所有这些文件并行运行 parse_file(...) 函数以减少程序执行时间并提高效率？如果是这样，由于程序没有使用 PySpark，是否需要进行大量代码修改才能使其并行化？

所以程序的功能是这样的

# ... some other codes
attributes_table = ....
for obj in gcp_bucket.objects(path):
    if obj.key.endswith('sys_data.txt'):
        #....some other codes
        file_data = (d for d in obj.download().decode('utf-8').split('\n'))
        parse_file(file_data, attributes_table)
        #....some other codes ....

如何使用 PySpark 并行化这部分，而不是一次使用 for 循环遍历文件？

Answer 1

感谢您提出问题。

我建议根据您的 gcp_bucket.objects(path) 创建一个 RDD。

你有你的 SparkContext，所以创建 RDD 应该很简单：
my_rdd = sc.parallelize(gcp_bucket.objects(path)。

对于外行来说，约定是将 SparkContext 赋值给变量sc。你的 for 循环的内容必须放入一个函数中，我们称之为my_function。你现在已经拥有了所有的碎片。

您的下一步将是这样映射您的函数：

results_dag = my_rdd.map(my_function)
results = results_dag.collect()

回想一下，Spark 执行惰性求值。这就是为什么我们需要在最后执行collect 操作。

其他一些建议。第一个是在 GCP 存储桶中的一小组对象上运行您的代码。了解时间安排。为了促进良好的编码实践，另一个建议是考虑将 for 循环中的操作进一步分解为额外的 RDD。您可以随时将它们链接在一起...

my_rdd = sc.parallelize(gcp_bucket.objects(path)
dag1 = my_rdd.map(function1)
dag2 = dag1.map(function2)
dag3 = dag2.map(function3)
results = dag3.collect()