0. 前言

Python 已逐渐成为数据分析/处理领域中的主要语言，这得益于 Python 丰富的第三方库，但是，这些库的设计并未在分布式上进行扩展。Dask 是为了原生地扩展这些 Python 库及其生态系统而开发的，它能够与现有的 Python 生态系统兼容，将其扩展到多核计算机和分布式集群中。

1. Dask 简介

Dask 是一个功能强大，可扩展且灵活的并行计算库，也是一个构建分布式应用程序的平台。它可以利用单个计算机的多核CPU，也可以扩展至多计算机群集。
在本节中，我们将学习如何使用 Dask 的 Sobel 滤波器的分布式实现在图像中进行分布式边缘检测。大型数据集通常无法将所有数据放入一个文件中，因此可能需要将它们分区并存储在不同的文件中，有时甚至可能存储在不同计算机中，为了模拟这种情况，我们将一个大图像分为四个较小的图像部分，并将这些部分作为单独的图像存储。然后，我们将介绍如何使用 Dask 读取这些图像块，并在每个图像上运行分布式边缘检测滤波器，最后组合图像以获取完整的边缘图像。
为了使用 Dask，我们首先需要使用 pip 命令进行安装：

$ pip install dask
$ pip install dask_image 

2. 使用 Dask 进行分布式图像处理

(1) 导入所有必需的库，模块和函数：

from skimage.io import imread, imsave
from skimage.color import rgb2gray
import numpy as np
import matplotlib.pylab as plt
import dask_image.imread
import dask.array
import dask_image.ndfilters
import os, glob
from IPython.display import display

(2) 实现函数 partion_image() 以分割图像，将图像分割为不重叠的图像块(默认情况下为 2x2=4 个图像)，然后，将这些文件保存到同一文件夹中：

def partion_image(imgfile, n_h=2, n_w=2, plot=True):
    im = imread(imgfile)
    h, w, _ = im.shape
    h_s, w_s = h // n_h, w // n_w
    k = 0
    for i in range(n_h):
        for j in range(n_w):
            imsave(imgfile[:-4] + '_part_{:03d}'.format(k) + imgfile[-4:], im[i*h_s:(i+1)*h_s, j*w_s:(j+1)*w_s, :])          
            k += 1
    if plot:
        k = 0
        plt.figure(figsize=(20,16))
        plt.subplots_adjust(0,0,1,1,0.05,0.05)
        for i in range(n_h):
            for j in range(n_w):
                im = plt.imread(imgfile[:-4] + '_part_{:03d}'.format(k) + imgfile[-4:])
                plt.subplot(n_h, n_w, k+1), plt.imshow(im), plt.title('image part-{}'.format(k+1), size=20)
                k += 1
        plt.show()

(3) 调用函数 partion_image() 将输入图像分块：

def plot_image(image):
    plt.figure(figsize=(20,20))
    plt.imshow(image, cmap='gray')
    plt.show()

imgfile = 'save.png'
partion_image(imgfile)

(4) 使用 dask_image的imread() 函数读取所有图像块：

filename_pattern = os.path.join('./', imgfile[:-4] + '_part_*' + imgfile[-4:])
partitioned_images = dask_image.imread.imread(filename_pattern)
print(partitioned_images)
# dask.array<_map_read_frame, shape=(4, 467, 825, 3), dtype=uint8, chunksize=(1, 467, 825, 3), chunktype=numpy.ndarray>

从以上输出可以看出，创建了一个 dask 数组，其形状为 (4, 467, 825, 3)，这意味着它包含四个图像帧，每个图像帧具有 467 行、825 列和 3 个颜色通道。

2.1 将 RGB 图像块转换为灰度图像块

将 RGB 图像块转换为灰度图像块的语法与将函数应用于多个图像或 Dask 块是相同的，这就是为什么 skimage.colors 中的 rgb2gray() 函数可以直接应用于 Dask 图像块。

(1) 绘制获得的第一个灰度图像块，Dask 能够使用现有的 scikit-image 函数将图像块转换为灰度：

result  = (rgb2gray(partitioned_images))
print(result.shape)
plot_image(result[0])

(2) 使用 dask.array.block() 函数，利用图像块创建组合图像，并绘制组合图像：

data = [result[i, ...] for i in range(result.shape[0])]
data = [data[i:i+2] for i in range(0, len(data), 2)]
combined_image = dask.array.block(data)
print(combined_image.shape)    
plot_image(combined_image)

2.2 使用分布式 Sobel 滤波器检测图像边缘

(1) 使用 Dask 自动为每个块创建计算图：

edges = dask_image.ndfilters.sobel(combined_image)
print(edges)
display(edges.visualize())

上图显示了 Dask 计算图，从图中可以明显的看出许多步骤能够并行完成。

(2) 最后，裁剪输出图像的像素值，使值在 [0,1] 范围内，并绘制边缘输出图像：

edges = np.clip(edges, 0, 1)
plot_image(edges)

小结

分布式处理可以将位于不同地点的、或具有不同功能的、或拥有不同数据的多台计算机通过通信网络连接起来，在控制系统的统一管理控制下，协调地完成大规模信息处理任务，利用分布式处理可以提高信息处理速度。在本节中，我们学习了如何使用 Dask 完成分布式图像处理，从而提高图像处理速度。

系列链接

Python图像处理【1】图像与视频处理基础
Python图像处理【2】探索Python图像处理库
Python图像处理【3】Python图像处理库应用
Python图像处理【4】图像线性变换
Python图像处理【5】图像扭曲/逆扭曲
Python图像处理【6】通过哈希查找重复和类似的图像
Python图像处理【7】采样、卷积与离散傅里叶变换
Python图像处理【8】使用低通滤波器模糊图像
Python图像处理【9】使用高通滤波器执行边缘检测
Python图像处理【10】基于离散余弦变换的图像压缩
Python图像处理【11】利用反卷积执行图像去模糊
Python图像处理【12】基于小波变换执行图像去噪
Python图像处理【13】使用PIL执行图像降噪
Python图像处理【14】基于非线性滤波器的图像去噪
Python图像处理【15】基于非锐化掩码锐化图像
Python图像处理【16】OpenCV直方图均衡化
Python图像处理【17】指纹增强和细节提取
Python图像处理【18】边缘检测详解
Python图像处理【19】基于霍夫变换的目标检测
Python图像处理【20】图像金字塔
Python图像处理【21】基于卷积神经网络增强微光图像
Python图像处理【22】基于卷积神经网络的图像去雾