无法使用 pyspark 同时[并行]读取图像答案

【问题标题】：Unable to read images simultaneously [in parallels] using pyspark无法使用 pyspark 同时[并行]读取图像
【发布时间】：2021-05-26 13:25:32
【问题描述】：

我在一个目录中有 10 个 jpeg 图像。我想使用 pyspark 同时阅读所有这些内容。我尝试如下。

from PIL import Image


from pyspark import SparkContext, SparkConf    

conf = SparkConf()
spark = SparkContext(conf=conf)       

files = glob.glob("E:\\tests\\*.jpg")

files_ = spark.parallelize(files)    

arrs = []

for fi in files_.toLocalIterator():      

    im = Image.open(fi)
    data = np.asarray(im)
    arrs.append(data)

img = np.array(arrs)    
print (img.shape)

代码无误结束，打印出img.shape；但是，它没有并行运行。你能帮帮我吗？

【问题讨论】：

你试过二进制文件数据源（spark.apache.org/docs/latest/sql-data-sources-binaryFile.html）吗？
在您通知我如下bds = spark.read.format("binaryFile").option("pathGlobFilter", "*.jpg").load("E:\\tests\\") print (bds)后，我尝试了。但是，它以错误结束，Exception: Java gateway process exited before sending its port number
@thunder 通常当我有这个异常是因为我的代码没有看到 JAVA_HOME 环境变量。检查是否是您的情况，然后重试
@Kafels 我已经在 C:\JAVA 中安装了jdk-16.0.1_windows-x64_bin 并将其设置为JAVA_HOME environment variable。
@thunder 嗯，对我来说唯一有效的版本是 Java 8 版本

标签： apache-spark pyspark parallel-processing python-imaging-library

【解决方案1】：

您可以使用rdd.map并行加载和转换图片，然后将rddcollect转换成Python列表：

files = glob.glob("E:\\tests\\*.jpg")

file_rdd = spark.parallelize(files)

def image_to_array(path):
    im = Image.open(path)
    data = np.asarray(im)
    return data

array_rdd = file_rdd.map(lambda f: image_to_array(f))
result_list = array_rdd.collect()

result_list 现在是一个包含 10 个元素的列表，每个元素都是一个 numpy.ndarray。

image_to_array 函数将在不同的 Spark 执行器上并行执行。如果你有一个多节点的 Spark 集群，你必须确保所有节点都可以访问E:\\tests\\。

收集完数组后，可以继续处理

img = np.array(result_list, dtype=object)

【讨论】：

你将如何开始 SparkSession 或 SparkContext 来完成这项任务，我正在努力解决它......
我正在使用spark = SparkSession.builder.appName("test").getOrCreate().sparkContext，但您在问题中给出的代码也可以使用

【解决方案2】：

我的解决方案与 werner 的想法相同，但我只使用了 spark 库：

from pyspark.ml.image import ImageSchema
import numpy as np


df = (spark
      .read
      .format("image")
      .option("pathGlobFilter", "*.jpg")
      .load("your_data_path"))

df = df.select('image.*')

# Pre-caching the required schema. If you remove this line an error will be raised.
ImageSchema.imageFields

# Transforming images to np.array
arrays = df.rdd.map(ImageSchema.toNDArray).collect()

img = np.array(arrays)
print(img.shape)

【讨论】：