将图像加载到 Dask Dataframe

Question

我有一个 dask 数据框，其中包含列中的图像路径（称为 img_paths）。在接下来的步骤中，我想做的是使用这些图像路径将图像加载到另一列（称为 img_loaded），然后应用一些预处理功能。

然而，在加载（或图像读取）过程中，我总是得到不同的结果，包括一次延迟包装 imread 函数，另一次正确加载图像（我可以看到数组）和其余的次：FileNotFoundError.

除了下面的示例，我还使用了 map_partitions 函数，但我也得到了类似的输出，只是没有数组。最后，我想使用 map_partitions 函数而不是 apply 函数。

以下是我的代码和问题描述：

import pandas as pd
import dask
import dask.dataframe as dd
from skimage.io import imread

imgs = ['https://cdn.sstatic.net/Sites/Whosebug/company/img/logos/so/so-logo.png?v=9c558ec15d8a'] * 42

# create a pandas dataframe using image paths
df = pd.DataFrame({"img_paths": imgs})

# convert it into dask dataframe
ddf = dd.from_pandas(df, npartitions=2)

# convert imread function as delayed
delayed_imread = dask.delayed(imread, pure=True)

第一次尝试：使用 lambda 函数并对每个单元格应用延迟 imread

ddf["img_loaded"] = ddf.images.apply(lambda x: delayed_imread(x))
ddf.compute()

这里我得到的是在使用 compute() 方法时包装延迟的 imread 函数。我不懂为什么？以下是输出：

第二次尝试：不使用 lambda 函数

ddf["img_loaded"] = ddf.images.apply(delayed_imread)
ddf.compute()

成功了！至少，我可以将加载的图像视为数组。但是，我真的不明白为什么？为什么这与第一个解决方案（即使用 lambda 函数）不同 以下是输出：

第三次尝试：with/without 使用 lambda 函数而不使用延迟 imread。

ddf["load"] = ddf.images.apply(imread) # or, lambda x: imread(x)
ddf.compute()

在这里，再次只是为了实验，我没有使用延迟的 imread 函数，而是简单地使用 skimage.io.imread 函数。而且，我尝试过使用和不使用 lambda 函数。每次，我得到FileNotFoundError。我没有得到这个。为什么使用非延迟imread函数找不到图片路径（虽然是正确的）？

除了Ronald的回答，map_partitions函数的使用方法：

ddf["img_loaded"] = ddf.map_partitions(lambda df: df.images.apply(lambda x: imread(x)), meta=("images", np.uint8)).compute()
ddf.compute()

Answer 1

解决方案

import pandas as pd
import dask
import dask.dataframe as dd
import numpy as np
from skimage.io import imread

imgs = ['https://cdn.sstatic.net/Sites/Whosebug/company/img/logos/so/so-logo.png?v=9c558ec15d8a'] * 4

# create a pandas dataframe using image paths
df = pd.DataFrame({"img_paths": imgs})

# convert it into dask dataframe
ddf = dd.from_pandas(df, npartitions=2)

# convert imread function as delayed
delayed_imread = dask.delayed(imread, pure=True)

# give dask information about the function output type
ddf['img_paths'].apply(imread, meta=('img_loaded', np.uint8)).compute()

# OR turn it into dask.dealayed, which infers output type `object`
ddf['img_paths'].apply(delayed_imread).compute()

解释

如果您尝试应用 print 函数，无需计算，您会看到代码 FileNotFoundError 的原因：ddf.images.apply(imread).compute()

ddf['img_paths'].apply(print)

输出：

> foo
> foo

当您将 apply 函数添加到图形中时，Dask 通过它运行字符串 foo 以推断输出类型 => imread 正在尝试打开名为 foo.

为了更好地理解，我鼓励您尝试：

ddf.apply(print, axis=1)

并尝试预测打印的内容。

.compute()

之后的延迟单元格

原因是 apply 需要一个函数引用，然后调用它。通过创建调用延迟函数的 lambda 函数，您基本上就是 double-delaying 您的函数。