当使用binaryFiles时，mapPartitions何时加载到内存中？

Question

我正在使用PySpark对图像应用一个经过训练的深度学习模型，并关注如何使用我目前的方法来扩展内存使用。由于经过训练的模型需要一段时间才能加载，所以我使用类似于以下代码的代码处理每个工作人员上的大量图像：

def run_eval(file_generator):
    trained_model = load_model()
    results = []
    for file in file_generator:
        # "file" is a tuple: [0] is its filename, [1] is the byte data
        results.append(trained_model.eval(file[1]))
    return(results)

my_rdd = sc.binaryFiles('adl://my_file_path/*.png').repartition(num_workers)
results = my_rdd.mapPartitions(run_eval)
results.collect()

如上所述，这些文件存储在一个相关的HDFS文件系统中(具体来说，是一个Azure Data )，可以通过SparkContext访问。

我的主要问题是：

• 何时将图像数据加载到内存中？
  • 当生成器增加(“刚好及时”)时，是否加载了每个映像的数据？
  • 在工作人员启动之前，是否加载了整个分区的所有图像数据？

​

• 是head节点负责从这个关联的文件系统加载数据(可能会造成瓶颈)，还是工作人员从它加载自己的数据？

也感谢你的建议，在哪里可以找到这些主题的深度覆盖。

Answer 1

何时将图像数据加载到内存中？

• 当生成器增加(“刚好及时”)时，是否加载了每个映像的数据？

实际上，给定您的代码，它必须加载不止一次。首先由JVM访问它，然后转换成Python类型。在此之后，将发生洗牌，并再次加载数据。每个进程都是延迟的，所以加载不是问题。

所以你必须问自己的第一个问题是你是否真的要洗牌。binaryFiles有minPartitions参数，可以用来控制分区的数量。

另一个问题是非懒惰的results list。使用生成器表达式更有意义：

def run_eval(file_generator):
    trained_model = load_model()
    for file in file_generator:
        yield trained_model.eval(file[1])

是head节点负责从这个关联的文件系统加载数据(可能会造成瓶颈)，还是工作人员从它加载自己的数据？

不涉及中央处理。每个执行者进程(Python) /线程(JVM)将加载数据集中自己的部分。