从 JPEG 创建 Spark 对象并在非转换函数上使用 spark_apply（）

问：

通常，当人们想要在自定义函数（即**非转换函数）上使用时，他们会将它们放在 spark_apply（） 中。但是，我只遇到过一些示例，其中单个本地数据帧是 or 到远程数据源，然后在其上使用。举个例子，仅供说明之用：sparklyrcopy_to()spark_read_csv()spark_apply()

library(sparklyr)
sc <- spark_connect(master = "local")

n_sim = 100
iris_samps <- iris %>% dplyr::filter(Species == "virginica") %>%
  sapply(rep.int, times=n_sim) %>% cbind(replicate = rep(1:n_sim, each = 50)) %>% 
  data.frame() %>%
  dplyr::group_by(replicate) %>%
  dplyr::sample_n(50, replace = TRUE)

iris_samps_tbl <- copy_to(sc, iris_samps)

iris_samps_tbl %>% 
  spark_apply(function(x) {mean(x$Petal_Length)}, 
    group_by = "replicate") %>%
  ggplot(aes(x = result)) + geom_histogram(bins = 20) + ggtitle("Histogram of 100 Bootstrapped Means using sparklyr")

因此，只要数据驻留在 Spark 对象中，就可以将其用于来自或包的任何范围的非转换函数。CRANBioconductor

当我读到 SparkR 可以通过 Java 库中的 ImageIO 将压缩图像（.jpeg、.png 等）加载到原始图像表示中时，我提出了一个针对图像的特定问题 - 似乎也可以做到这一点。.jpegsparklyr

RsimMosaic::composeMosaicFromImageRandom(inputImage, outputImage, pathToTilesLibrary)函数获取输入图像和用于创建照片马赛克的图块路径，并输出图像（此处为示例）。

如果这个函数只拍摄一张图像，并且我知道如何将其转换为火花对象，我可能会想象该命令将如下所示：.但是，此函数正在获取 30,000 张图像的路径。composeMosaicFromImageRandom(inputImage, outputImage, spark_obj)

如何从这些图块 （） 的路径创建 30,000 个 Spark 对象，然后使用此功能？.jpegs

如果底层代码确实需要修改，我曾经提供源代码：jimhester/lookup

function (originalImageFileName, outputImageFileName, imagesToUseInMosaic, 
        useGradients = FALSE, removeTiles = TRUE, fracLibSizeThreshold = 0.7, 
        repFracSize = 0.25, verbose = TRUE) 
{
        if (verbose) {
                cat(paste("\n ------------------------------------------------ \n"))
                cat(paste("    R Simple Mosaic composer - random version   \n"))
                cat(paste(" ------------------------------------------------ \n\n"))
        }
        if (verbose) {
                cat(paste("    Creating the library... \n"))
        }
        libForMosaicFull <- createLibraryIndexDataFrame(imagesToUseInMosaic, 
                saveLibraryIndex = F, useGradients = useGradients)
        libForMosaic <- libForMosaicFull
        filenameArray <- list.files(imagesToUseInMosaic, full.names = TRUE)
        originalImage <- jpeg::readJPEG(filenameArray[1])
        xTileSize <- dim(originalImage[, , 1])[1]
        yTileSize <- dim(originalImage[, , 1])[2]
        if (verbose) {
                cat(paste("    -- Tiles in the Library : ", length(libForMosaic[, 
                        1]), "\n"))
                cat(paste("    -- Tile dimensions : ", xTileSize, " x ", 
                        yTileSize, "\n"))
        }
        if (verbose) {
                cat(paste("\n"))
                cat(paste("    Reading the original image... \n"))
        }
        originalImage <- jpeg::readJPEG(originalImageFileName)
        xOrigImgSize <- dim(originalImage[, , 1])[1]
        yOrigImgSize <- dim(originalImage[, , 1])[2]
        if (verbose) {
                cat(paste("    -- Original image dimensions : ", xOrigImgSize, 
                        " x ", yOrigImgSize, "\n"))
                cat(paste("    -- Output image dimensions : ", ((xOrigImgSize - 
                        2) * xTileSize), " x ", ((yOrigImgSize - 2) * yTileSize), 
                        "\n"))
        }
        if (verbose) {
                cat(paste("\n"))
                cat(paste("    Computing the mosaic... \n"))
        }
        outputImage <- array(dim = c(((xOrigImgSize - 2) * xTileSize), 
                ((yOrigImgSize - 2) * yTileSize), 3))
        removedList <- c()
        l <- 1
        pCoord <- matrix(nrow = ((xOrigImgSize - 2) * (yOrigImgSize - 
                2)), ncol = 2)
        for (i in 2:(xOrigImgSize - 1)) {
                for (j in 2:(yOrigImgSize - 1)) {
                        pCoord[l, 1] <- i
                        pCoord[l, 2] <- j
                        l <- l + 1
                }
        }
        npixels <- length(pCoord[, 1])
        for (i in 1:npixels) {
                idx <- round(runif(1, 1, length(pCoord[, 1])))
                pixelRGBandNeigArray <- computeStatisticalQuantitiesPixel(pCoord[idx, 
                        1], pCoord[idx, 2], originalImage, useGradients)
                tileFilename <- getCloseMatch(pixelRGBandNeigArray, 
                        libForMosaic)
                startI <- (pCoord[idx, 1] - 2) * xTileSize + 1
                startJ <- (pCoord[idx, 2] - 2) * yTileSize + 1
                outputImage[startI:(startI + xTileSize - 1), startJ:(startJ + 
                        yTileSize - 1), ] <- jpeg::readJPEG(tileFilename)
                if (removeTiles) {
                        libForMosaic <- removeTile(tileFilename, libForMosaic)
                        removedList <- c(removedList, tileFilename)
                        if (length(libForMosaic[, 1]) < (fracLibSizeThreshold * 
                                length(libForMosaicFull[, 1]))) {
                                idxs <- runif(round(0.25 * length(libForMosaicFull[, 
                                        1])), 1, length(removedList))
                                for (ii in 1:length(idxs)) {
                                        libForMosaic <- addBackTile(removedList[idxs[ii]], 
                                                libForMosaic, libForMosaicFull)
                                }
                                removedList <- removedList[-idxs]
                        }
                }
                if (length(pCoord[, 1]) > 2) {
                        pCoord <- pCoord[-idx, ]
                }
        }
        if (verbose) {
                cat(paste("\n"))
                cat(paste("    Done!\n\n"))
        }
        jpeg::writeJPEG(outputImage, outputImageFileName)
}

请注意：我第一次尝试加速这段代码是 1） use 查找瓶颈（即 for 循环）2） 在 for 循环上使用包。这导致代码速度变慢，这表明我的并行化水平太低。据我所知，更多的是关于分布式计算，而不是并行化它，所以也许这可以工作。profvisforeachsparklyr