我可以使用 Spark 进行自定义计算吗？

Posted 2023-04-15

【中文标题】我可以使用 Spark 进行自定义计算吗？

【英文标题】：Can I use Spark for custom computation?

【发布时间】：2020-09-30 14:19:34

【问题描述】：

我有一些（200 多个）大型 zip 文件（一些 >1GB）应该使用 Python 地理和图像处理库进行解压缩和处理。结果将作为新文件写入 FileStore，稍后用于 Databricks 中的 ML 任务。

如果我想利用 Spark 集群的处理能力，一般方法是什么？我正在考虑将文件名添加到 DataFrame，并使用用户定义的函数通过 Select 或类似方法处理它们。我相信我应该能够在集群上并行运行，工作人员将只获取文件名，然后在本地加载文件。

这是合理的，还是我应该走一些完全不同的方向？

更新 - 或者可能是这样的：

zipfiles = ...

def f(x):
  print("Processing " + x)

spark = SparkSession.builder.appName('myApp').getOrCreate()
rdd = spark.sparkContext.parallelize(zipfiles)
rdd.foreach(f)

更新 2： 对于任何这样做的人。由于默认情况下 Spark 将保留几乎所有可用内存，因此您可能必须使用以下设置来减少它： spark.executor.memory 1g 或者您可能会很快耗尽工作人员的内存。

【问题讨论】：

我认为这行不通，因为文件不在工作人员身上，因此您将无法找到和处理它们。 Spark用于数据处理，数据组织在数据框中。你想要做的是多线程，有大量的 Python 库。但火花不是其中之一。

master上的dbfs里的文件我都可以枚举出来，相信worker也应该可以加载了吧？

我想利用集群工作者，而不仅仅是在一台机器上运行多线程。

根据我的经验，它不会起作用，即使你以某种方式使它起作用（我希望那时看到）它也不会有效，因为 spark 不是为此而生的。只需看看数据框的可用功能。他们都不处理文件。这是一个数据工程框架。使用纯 python 进程会更好。此外，了解 map/reduce 的工作原理应该会让您明白以这种方式处理文件是不可能的。

嗨史蒂文 - 看起来这确实工作得很好。请参阅我对问题的更新。我的用户定义函数从 DBFS 加载文件并处理它们，然后将结果写回 DBSF。并发由 parallelize() 方法控制。

【参考方案1】：

是的，您可以将 Spark 用作通用并行处理引擎，提供或接受一些序列化问题。例如，在一个项目中，我使用 spark 并行扫描许多布隆过滤器，以及布隆过滤器返回正值的随机访问索引文件。对于此类量身定制的解决方案，您很可能需要使用 RDD api。

【讨论】：

到目前为止它似乎工作得很好，谢谢。我正在使用 RDD 并进行并行化。