python - 从 PySpark 中的工作节点访问 ADLS 上的二进制文件的最有效方法？

Question

我已经部署了一个 Azure HDInsight 集群，它对 Azure Data Lake Store 上的所有目录都具有 rwx 权限，该存储也用作其存储帐户。在头节点上，我可以加载例如使用如下命令从 ADLS 获取图像数据:

my_rdd = sc.binaryFiles('adl://{}.azuredatalakestore.net/my_file.png')

工作人员无权访问 SparkContext 的 binaryFiles() 函数。我可以使用 azure-datalake-store Python SDK 来加载文件，但这似乎要慢得多。我假设是因为它没有意识到集群和 ADLS 之间关联的任何好处。

有没有更快的方法从工作人员上的关联 ADLS 加载文件？

如果需要更多上下文:

我正在使用 PySpark 将经过训练的深度学习模型应用于大量图像。由于模型需要很长时间才能加载，我的理想是:

• 向每个工作人员发送要处理的图像 URI 的部分列表(通过将 mapPartition() 应用于包含完整列表的 RDD)
• 让工作人员一次加载一张图像的数据，以便对模型进行评分
• 返回图像集的模型结果

由于我不知道如何在工作人员上有效地加载图像，目前我最好的选择是对包含图像字节数据的 RDD 进行分区，这(我认为)是内存效率低下的，并且会造成瓶颈头节点执行所有数据加载。

Answer 1

HDInsight 集群的主存储仅作为 HDFS 根可用。

hdfs dfs -ls /user/digdug/images/
Found 3 items
-rw-r--r--   1    digdug supergroup       4957 2017-01-24 07:59 /user/digdug/images/a.png
-rw-r--r--   1    digdug supergroup       4957 2017-01-24 07:59 /user/digdug/images/b.png
-rw-r--r--   1    digdug supergroup       1945 2017-01-24 08:01 /user/digdug/images/c.png

在 pyspark 中:

rdd = sc.binaryFiles("/user/digdug/images")

def f(iterator):
    sizes = []
    for i in iterator:
        sizes.append(len(i[1]))
    return sizes

rdd.mapPartitions(f).collect()

输出:

[4957, 4957, 1945]