使用Apache Spark和NVIDIA GPU加速深度學(xué)習(xí)

隨著人們對(duì)深度學(xué)習(xí)（ deep learning ， DL ）興趣的日益濃厚，越來越多的用戶在生產(chǎn)環(huán)境中使用 DL 。由于 DL 需要強(qiáng)大的計(jì)算能力，開發(fā)人員正在利用 gpu 來完成他們的訓(xùn)練和推理工作。

最近，為了更好地統(tǒng)一 Spark 上的 DL 和數(shù)據(jù)處理，作為 Apache Spark 的一項(xiàng)重大舉措的一部分， GPU 成為 Apache Spark 3.0 中的一種可調(diào)度資源。 Spark 將這些資源請(qǐng)求傳遞給底層集群管理器。因?yàn)檫@允許您大規(guī)模地運(yùn)行分布式推理，所以它可以幫助加速大數(shù)據(jù)管道以利用 DL 應(yīng)用程序。

在 Apache Spark 3. 0 之前，使用 gpu 是很困難的。您必須手動(dòng)將 GPU 設(shè)備分配給 Spark 作業(yè)，并對(duì)每個(gè)執(zhí)行器或任務(wù)的所有配置進(jìn)行硬編碼，以便在一臺(tái)機(jī)器上使用不同的 GPU 。因?yàn)?apachehadoop3 。 1yarn 集群管理器允許不同機(jī)器之間的 GPU 協(xié)調(diào)， Apache Spark 現(xiàn)在可以與它一起工作，幫助將設(shè)備安排傳遞給不同的任務(wù)。提交帶有 GPU 資源配置發(fā)現(xiàn)腳本的應(yīng)用程序后， Spark 將處理任務(wù)之間如何共享 GPU 的分配和協(xié)調(diào)。

在本教程中，我們將演示如何創(chuàng)建 GPU 機(jī)器集群，并使用 Apache Spark 和 Amazon EMR 上的 深層 Java 庫（ DJL ） 來利用 Scala 中的大規(guī)模圖像分類。 DJL 現(xiàn)在提供了一個(gè)基于 GPU 的深入學(xué)習(xí) Java 包，該包被設(shè)計(jì)成可以在 Spark 中順利工作。

如果您對(duì) Scala 和 Java 感興趣，或者正在尋找將 Java 中的 DL 集成到大數(shù)據(jù)管道中的解決方案， DJL 提供了一個(gè)可行的解決方案。由于 Python 是 DL 最常用的語言，而 Java 是企業(yè)開發(fā)人員和數(shù)據(jù)工程師最流行的語言， DJL 的目標(biāo)是深入學(xué)習(xí)， Java 開發(fā)人員可以使用熟悉的概念和直觀的 API 訪問的開源工具。 DJL 是建立在現(xiàn)代深度學(xué)習(xí)框架（ TensorFlow 、 PyTorch 、 Apache MXNet 等）之上的。您可以輕松地使用 DJL 來訓(xùn)練您的模型，或者從各種引擎部署您喜愛的模型，而無需進(jìn)行任何額外的轉(zhuǎn)換。

設(shè)置 Spark 應(yīng)用程序

首先，導(dǎo)入 Spark 依賴項(xiàng)。 Spark SQL 和 ML 庫用于存儲(chǔ)和處理映像， Spark 依賴項(xiàng)僅在編譯時(shí)使用，并且由于在運(yùn)行時(shí)提供，因此在打包時(shí)被排除在外。當(dāng)所有東西都打包好后，。 jar 任務(wù)將它們排除在外。

configurations {
    exclusion
}
dependencies {
    implementation "org.apache.spark:spark-sql_2.12:3.0.1"
    implementation "org.apache.spark:spark-mllib_2.12:3.0.1"
    implementation "org.apache.hadoop:hadoop-hdfs:2.7.4"

    exclusion "org.apache.spark:spark-sql_2.12:3.0.1"
    exclusion "org.apache.spark:spark-mllib_2.12:3.0.1"
    exclusion "org.apache.hadoop:hadoop-hdfs:2.7.4"}
}
jar {
    from {
        (configurations.runtimeClasspath - configurations.exclusion).collect {
            it.isDirectory() ? it : zipTree(it)
        }
    }
}

接下來，導(dǎo)入與 DJL 相關(guān)的依賴項(xiàng)。您使用 DJL 和 PyTorch 包。它們提供了 DJL 的核心特性，并加載了一個(gè) DL 引擎來運(yùn)行以進(jìn)行推斷。此外，您還可以使用 pytorch-native-cu101 在具有 CUDA 10 。 1 的 GPU 上運(yùn)行。

    implementation platform("ai.djl:bom:0.8.0")
    implementation "ai.djl:api"
    runtimeOnly "ai.djl.pytorch:pytorch-model-zoo"
    runtimeOnly "ai.djl.pytorch:pytorch-native-cu101::linux-x86_64"

加載模型

要在 DJL 中加載模型，請(qǐng)?zhí)峁┏休d模型的 URL （ file ：//， hdfs ：//， s3 ：//， https ：//）。從該 URL 下載并導(dǎo)入模型。 DJL 還提供了一個(gè)強(qiáng)大的動(dòng)物園模型。 zoo 模型允許您管理預(yù)先訓(xùn)練的模型，并在一行中加載它們。內(nèi)置的動(dòng)物園模型目前支持 70 多個(gè)預(yù)先訓(xùn)練和準(zhǔn)備使用的模型，這些模型來自 GluonCV 、 HuggingFace 、 TorchHub 和 Keras 。

def loadModel(device : Device): ZooModel[Row, Classifications] = {
    val modelUrl = "https://alpha-djl-demos.s3.amazonaws.com/model/djl-blockrunner/pytorch_resnet18.zip?model_name=traced_resnet18"
    val criteria = Criteria.builder
      .setTypes(classOf[Row], classOf[Classifications])
      .optModelUrls(modelUrl)
      .optTranslator(new MyTranslator())
      .optProgress(new ProgressBar)
      .optDevice(device)
      .build()
     ModelZoo.loadModel(criteria)
}

這里的輸入類型是 Spark SQL 中的 Row 。輸出類型是分類結(jié)果。 MyTranslator 函數(shù)執(zhí)行預(yù)處理和后處理工作。加載的模型是來自 torchvision 的預(yù)訓(xùn)練 PyTorch ResNet18 模型。

主要邏輯

在下面的代碼示例中， downloadImages 函數(shù)下載演示圖像并將其存儲(chǔ)在 Hadoop 文件系統(tǒng)（ hdfs ）。接下來， spark.read.format（“image”） 函數(shù)使用 Spark 圖像數(shù)據(jù)源 將圖像文件從 HDFS 加載到 Spark DataFrame 中。在此步驟之后， mapPartition 獲取 GPU 信息。如代碼示例所示， TaskContext.resources（）（“gpu”） 函數(shù)存儲(chǔ)為此分區(qū)分配的 GPU 。這可確保單個(gè)設(shè)備上的所有 GPU 都得到正確使用。將模型加載到指定的 GPU 后， predictor.predict（row） 返回 Spark DataFrame 分區(qū)中圖像（行）的分類。

def main(args: Array[String]) {

    // download images
    val imagePath = downloadImages(new Path("hdfs:///images"))

    // Spark configuration
    val spark = SparkSession.builder()
      .appName("Image Classification")
      .config(new SparkConf())
      .getOrCreate()
    val df = spark.read.format("image").option("dropInvalid", true).load(imagePath)
    val result = df.select(col("image.*")).mapPartitions(partition => {
      val context = TaskContext.get()
      val gpu = context.resources()("gpu").addresses(0)
      val model = loadModel(Device.gpu(gpu.toInt))
      val predictor = model.newPredictor()
      partition.map(row => {
        predictor.predict(row).toString
      })
    })(Encoders.STRING)
    println(result.collect().mkString("
"))
}

把它包起來

運(yùn)行 。/gradlew jar 將所有內(nèi)容捆綁到一個(gè) jar 中，并在 Spark 集群中運(yùn)行。

使用多個(gè) GPU 設(shè)置 Spark 群集

由于 Amazon emr6 。 2 。 0 的發(fā)布， Spark 3 。 0 在所有 GPU 實(shí)例中都可用。

要設(shè)置 Spark 群集，請(qǐng)使用 AWS CLI 創(chuàng)建一個(gè)包含三個(gè)實(shí)例的 GPU 群集。要成功運(yùn)行該命令，必須將 myKey 更改為 EC2 密鑰名稱。如果預(yù)先配置了 --region 選項(xiàng)，也可以將其刪除。

aws emr create-cluster \
    --name "Spark cluster" \
    --release-label emr-6.2.0 \
    --region us-east-1 \
    --ebs-root-volume-size 50 \
    --applications Name=Hadoop Name=Spark \
    --ec2-attributes KeyName=myKey \
    --instance-type g3s.xlarge \
    --instance-count 3 \
    --use-default-roles \
    --configurations https://raw.githubusercontent.com/aws-samples/djl-demo/master/aws/emr-distributed-inference/image-classification-gpu/configurations.json 

您可以從 AWS 中提供的各種 GPU 實(shí)例中進(jìn)行選擇。此示例使用 g3s.xlarge 實(shí)例類型進(jìn)行測試目的群集設(shè)置的總運(yùn)行時(shí)間約為 10 – 15 分鐘。

執(zhí)行 Spark 作業(yè)

您可以在 EMR 控制臺(tái)上或從命令行運(yùn)行此駐車作業(yè)。

下面的命令告訴 Spark 運(yùn)行一個(gè) Yarn 集群，并設(shè)置一個(gè)腳本來查找不同設(shè)備上的 gpu 。每個(gè)任務(wù)的 GPU 數(shù)量設(shè)置為 0 。 5 ，這意味著兩個(gè)任務(wù)共享一個(gè) GPU 。您可能還需要相應(yīng)地設(shè)置 CPU 編號(hào)，以確保它們匹配。例如，如果您有一個(gè) 8 核 CPU ，并且將 spark.task.cpus 設(shè)置為 2 ，這意味著四個(gè)任務(wù)可以在一臺(tái)機(jī)器上并行運(yùn)行。要獲得最佳性能，請(qǐng)將 spark.task.resource.gpu.amount 設(shè)置為 0 。 25 。這允許四個(gè)任務(wù)共享同一個(gè) GPU 。這有助于最大限度地提高性能，因?yàn)?GPU 和 CPU 中的所有核心都已使用。如果沒有平衡的設(shè)置，一些內(nèi)核處于空閑狀態(tài)，這會(huì)浪費(fèi)資源。

spark-submit \
    --master yarn \
    --conf spark.executor.resource.gpu.discoveryScript=/usr/lib/spark/scripts/gpu/getGpusResources.sh \
    --conf spark.worker.resource.gpu.discoveryScript=/usr/lib/spark/scripts/gpu/getGpusResources.sh \
    --conf spark.task.resource.gpu.amount="0.5" \
    --conf spark.task.cpus=2 \
    --conf spark.executor.resource.gpu.amount=1 \
    --conf spark.worker.resource.gpu.amount=1 \
    --class com.examples.ImageClassificationExample \
    build/libs/image-classification-gpu-1.0-SNAPSHOT.jar

這個(gè)腳本大約需要 4-6 分鐘才能完成，您將得到一個(gè)打印輸出的推斷結(jié)果作為輸出。

摘要

在本教程中，您從頭開始構(gòu)建包，并將工作提交到 GPU 集群以執(zhí)行推理任務(wù)。嘗試對(duì)自己的應(yīng)用程序使用相同的設(shè)置。

關(guān)于作者

Qing Lan 是 AWS 深度學(xué)習(xí)工具包團(tuán)隊(duì)的 SDE 。他是 DJL 的合著者之一 （djl 。 ai 公司 ） 以及 apachemxnet 的 PPMC 成員。 2017 年畢業(yè)于哥倫比亞大學(xué)，獲計(jì)算機(jī)工程碩士學(xué)位， 2017 年暑期實(shí)習(xí)。他專注于分布式深度學(xué)習(xí)訓(xùn)練和推理領(lǐng)域。

Kong Zhao 是 NVIDIA 的主要解決方案架構(gòu)師。他提供技術(shù)思想領(lǐng)導(dǎo)和架構(gòu)指導(dǎo)。他為 NVIDIA 和 AWS 客戶進(jìn)行 PoC ，通過在云中開發(fā)、優(yōu)化和部署 GPU 加速解決方案來滿足他們的 AI 和 HPC 需求。他關(guān)注的核心領(lǐng)域是 GPU 相關(guān)的云架構(gòu)、 HPC 、機(jī)器學(xué)習(xí)和分析。此前， Kong 曾擔(dān)任 AWS 的高級(jí)解決方案架構(gòu)師、云交換的 atEquinix 架構(gòu)師以及 Cisco 的產(chǎn)品經(jīng)理。

Carol McDonald 是一位產(chǎn)品營銷經(jīng)理，專注于 Spark 和數(shù)據(jù)科學(xué)。 Carol 在很多方面都有經(jīng)驗(yàn)，包括技術(shù)營銷、軟件架構(gòu)和開發(fā)、培訓(xùn)、技術(shù)宣傳和開發(fā)人員外展。 Carol 編寫行業(yè)架構(gòu)、最佳實(shí)踐、模式、原型、教程、演示、博客文章、白皮書和電子書。她走遍世界各地，做演講和動(dòng)手實(shí)驗(yàn)；在銀行、醫(yī)療保險(xiǎn)和電信行業(yè)開發(fā)了復(fù)雜的、關(guān)鍵任務(wù)的應(yīng)用程序?？_爾擁有田納西大學(xué)計(jì)算機(jī)科學(xué)碩士學(xué)位和范德比爾特大學(xué)地質(zhì)學(xué)學(xué)士學(xué)位?？_爾精通英語、法語和德語。

審核編輯：郭婷