BigDL：Apache Spark 上的分布式深度学习

Intel

5.00/5 (1投票)

2017 年 4 月 19 日

CPOL

15分钟阅读

7069

本文介绍了 BigDL，演示了如何在各种平台上构建该库，并提供了 BigDL 的实际应用示例。

BigDL 是一个用于 Apache Spark* 的分布式深度学习库。使用 BigDL，您可以将深度学习应用程序编写为 Scala 或 Python* 程序，并利用可扩展的 Spark 集群的强大功能。本文介绍了 BigDL，展示了如何在各种平台上构建该库，并提供了 BigDL 的实际应用示例。

什么是深度学习？

深度学习是机器学习的一个分支，它使用算法对数据中的高级抽象进行建模。这些方法基于人工神经网络拓扑，并且可以随着更大的数据集而扩展。

图 1 探讨了传统方法和深度学习方法之间的根本区别。传统方法倾向于将特征提取作为训练机器学习模型的单一阶段，而深度学习通过在模型中创建特征提取器管道来引入深度。深度学习管道中的这些多个阶段通过分层特征提取提高了最终模型的整体预测准确性。

图 1. 深度学习作为特征提取器管道

什么是 BigDL？

BigDL 是一个用于 Spark 的分布式深度学习库，可以直接运行在现有的 Spark 或 Apache Hadoop* 集群之上。您可以将深度学习应用程序编写为 Scala 或 Python 程序。

丰富的深度学习支持。 BigDL 以 Torch 为模型，提供全面的深度学习支持，包括数值计算（通过 Tensor 和高级神经网络）；此外，您可以将预训练的 Caffe* 或 Torch 模型加载到 Spark 框架中，然后使用 BigDL 库对它们的数据运行推理应用程序。
高效的横向扩展。 BigDL 可以通过使用 Spark 以及 Spark 中同步随机梯度下降 (SGD) 和 all-reduce 通信的高效实现，高效地横向扩展以执行“大数据规模”的数据分析。
极高的性能。 为了获得高性能，BigDL 在每个 Spark 任务中使用 Intel® 数学核心库 (Intel® MKL) 和多线程编程。因此，它比在单个 Intel® Xeon® 处理器上开箱即用的开源 Caffe、Torch 或 TensorFlow（即，与主流图形处理单元相当）快了几个数量级。

什么是 Apache Spark*？

Spark 是由加州大学伯克利分校 AMPLab 开发的闪电般快速的分布式数据处理框架。Spark 可以独立运行，也可以在 Hadoop 之上的 YARN 集群模式下运行，或者在 Apache Mesos* 集群管理器中运行（图 2）。Spark 可以处理来自各种来源的数据，包括 HDFS、Apache Cassandra* 或 Apache Hive*。其高性能得益于通过内存持久化 RDD 或 DataFrame 进行内存处理的能力，而不是像传统的 Hadoop MapReduce 架构那样将数据保存到硬盘。

图 2. BigDL 在 Apache Spark* 栈中

为什么使用 BigDL？

如果您想编写深度学习程序，您可能会想使用 BigDL，因为

您想在数据所在的大数据 Spark 集群（例如，HDFS、Apache HBase* 或 Hive）上分析大量数据；
您想在您的大数据 (Spark) 程序或工作流中添加深度学习功能（训练或预测）；或者
您想使用现有的 Hadoop/Spark 集群来运行您的深度学习应用程序，然后您可以轻松地与其他人共享（例如，提取-转换-加载、数据仓库、特征工程、经典机器学习、图分析）。使用 BigDL 的一个不受欢迎的替代方案是，为了实现深度学习算法而引入另一个与 Spark 并行的分布式框架。

安装工具链

注意：本文档提供了全新安装的说明，假设您没有安装任何 Linux* 工具。如果您不是 Linux 新手，请随时跳过 Git 安装等基本步骤。

在 Oracle* VM VirtualBox 上安装工具链

要在 Oracle* VM VirtualBox 上安装工具链，请执行以下步骤：

在您计算机的 BIOS 设置中启用虚拟化（图 3）。
BIOS 设置通常在安全菜单下。您需要重新启动机器并进入 BIOS 设置菜单。

图 3. 在您计算机的 BIOS 中启用虚拟化。
确定您拥有 64 位还是 32 位计算机。
大多数现代计算机都是 64 位的，但为了确保万无一失，请查看“控制面板”>“系统和安全”小程序（图 4）。

图 4. 确定您的计算机是 32 位还是 64 位。

注意：您应该拥有超过 4 GB 的 RAM。您的虚拟机 (VM) 将运行 2 GB，但大多数 BigDL 示例（LeNet 除外）将无法正常工作（或根本无法工作）。本文认为 8 GB RAM 是最低要求。
从 VirtualBox 网站安装带有 GuestAdditions 或其他 VM 客户端的 VirtualBox。
图 5 中的 VM 配置已知可以与 BigDL 配合使用。

图5。

重要
- 为您的 VM 分配 35-50 GB 的硬盘空间，用于 Ubuntu*、Spark、BigDL 和所有深度学习模型。
- 分配硬盘空间时，选择“动态”分配，这允许您稍后扩展分区（即使它并不简单）。如果您选择“静态”，然后磁盘空间用完，您唯一的选择将是擦除整个安装并重新开始。
从 Ubuntu 下载页面安装 Ubuntu Desktop。
下载完成后，将您的 VM 指向下载的文件，以便您可以从中启动。（有关说明，请参阅 Ubuntu FAQ）
sudo apt-get update sudo apt-get upgrade

从现在开始，所有安装都将在 Ubuntu 或您的 Linux 版本中进行；除了明确说明的情况外，VM 或原生 Linux 的安装说明应该相同。

如果您在访问 Internet 时遇到问题，您可能需要设置代理。（虚拟专用网络，尤其是 Cisco AnyConnect，以修改 VM 的代理设置而闻名。）图 6 中的代理设置对于 VirtualBox 版本 5.1.12 和 64 位 Ubuntu 是有效的。

图 6. Oracle* VM VirtualBox 版本 5.1.12 和 64 位 Ubuntu* 系统的代理设置
使用以下命令安装 Java*
sudo add-apt-repository ppa:webupd8team/java sudo apt-get update sudo apt-get install oracle-java8-installer
验证 Java 版本
java –version
从 Scala 下载页面安装 Scala 版本 2.11。
下载 Scala 时，使用 Debian* 文件格式，它默认下载到 Downloads 文件夹。在文件浏览器中打开 Scala，然后单击它。

sudo apt-get update
从 Spark 下载页面安装 Spark。
以下说明适用于 Spark-1.6。如果您安装了其他版本，请将 `1.6.x` 替换为您选择的版本。

a. 替换 Spark 为您下载的版本

$ cd Downloads $ tar -xzf spark-1.6.1-bin-hadoop2.6.tgz

b. 将文件移动到正确的位置

$ sudo mkdir /usr/local/spark $ sudo mv spark-1.6.1-bin-hadoop2.6 /usr/local/spark

c. 测试安装

$ cd /usr/local/spark $ cd spark-2.1.0-bin-hadoop2.6 $ ./bin/spark-shell

d. 安装 Git

$ sudo apt-get install git

下载 BigDL

BigDL 可通过 GitHub* 使用 Git 源代码控制工具获取。按如下方式克隆 BigDL Git 存储库：

$ git clone https://github.com/intel-analytics/BigDL.git

完成后，您会看到一个名为 `BigDL` 的新子目录，其中包含 BigDL 存储库。您还必须安装 Apache Maven，如下所示，以编译 Scala 代码。

$ sudo apt-get install maven

构建 BigDL

本节介绍如何在您的 Linux 发行版上下载和构建 BigDL。构建 BigDL 的先决条件是：

Java 8（以获得最佳性能）；
Scala 2.10 或 2.11（如果您计划使用 Spark 2.0 或更高版本，则需要 Scala 2.11）；
Maven 3；以及
Git。

Apache Maven 环境

您需要 Maven 3 来构建 BigDL。您可以从 Maven 网站下载。

安装 Maven 3 后，按如下方式设置环境变量 `MAVEN_OPTS`：

$ export MAVEN_OPTS="-Xmx2g -XX:ReservedCodeCacheSize=512m"

使用 Java 7 编译时，必须添加 `-XX:MaxPermSize=1G` 选项。

构建

强烈建议您使用 make-dist.sh 脚本 构建 BigDL。

下载完成后，使用以下命令构建 BigDL：

$ bash make-dist.sh

之后，您可以找到一个 `dist` 文件夹，其中包含运行 BigDL 程序所需的所有文件。`dist` 中的文件包括：

dist/bin/bigdl.sh。使用此脚本设置正确的环境变量并启动 BigDL 程序。
dist/lib/bigdl-0.1.0-SNAPSHOT-jar-with-dependencies.jar。这个 Java 归档 (JAR) 包包含除 Spark 之外的所有依赖项。
dist/lib/bigdl-0.1.0-SNAPSHOT-jar-with-dependencies-and-spark.jar。这个 JAR 包包含包括 Spark 在内的所有依赖项。

替代方法

如果您是为 Spark 2.0 构建，您应该修改 `bash` 调用以使用 Scala 2.11。要构建 Spark 2.0（默认使用 Scala 2.11），请将 `-P spark_2.0` 传递给 `make-dist.sh` 脚本：

$ bash make-dist.sh -P spark_2.0

强烈建议您在运行 Spark 2.0 时使用 Java 8。否则，您可能会遇到性能不佳的情况。

默认情况下，`make-dist.sh` 对 Spark 版本 1.5.x 或 1.6.x 使用 Scala 2.10，对 Spark 2.0 使用 Scala 2.11。要覆盖默认行为，请根据需要将 `-P scala_2.10` 或 `-P scala_2.11` 传递给 `make-dist.sh`。

入门

在运行 BigDL 程序之前，您必须首先进行一些设置。本节确定了入门步骤。

设置环境变量

为了实现高性能，BigDL 使用 Intel MKL 和多线程编程。因此，您必须首先通过运行 `PATH_To_BigDL/scripts/bigdl.sh` 中的提供脚本来设置环境变量，如下所示：

$ source PATH_To_BigDL/scripts/bigdl.sh

或者，您可以使用 `PATH_To_BigDL/scripts/bigdl.sh` 来启动您的 BigDL 程序。

使用交互式 Scala Shell

您可以使用交互式 Scala Shell 来试验 BigDL 代码。要做到这一点，运行：

$ scala -cp bigdl-0.2.0-SNAPSHOT-jar-with-dependencies-and-spark.jar

然后，您可以看到类似以下内容：

Welcome to Scala 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_91).
Type in expressions for evaluation. Or try :help.

scala>

例如，要试验 BigDL 中的 Tensor 应用程序编程接口 (API)，您可以尝试：

scala> import com.intel.analytics.bigdl.tensor.Tensor
import com.intel.analytics.bigdl.tensor.Tensor

scala> Tensor[Double](2,2).fill(1.0)
res9: com.intel.analytics.bigdl.tensor.Tensor[Double] =
1.0     1.0
1.0     1.0
[com.intel.analytics.bigdl.tensor.DenseTensor of size 2x2]

有关 BigDL API 的更多详细信息，请参阅 BigDL 编程指南。

运行本地 Java* 程序（单机模式）

您可以将 BigDL 程序，例如 VGG 模型训练，作为在单节点（机器）上运行的本地 Java 程序来运行。要做到这一点，请执行以下步骤：

从 CIFAR-10 数据集页面下载 CIFAR-10 数据。
请记住选择二进制版本。
使用 `bigdl.sh` 脚本将示例作为本地 Java 程序启动：
```
./dist/bin/bigdl.sh -- \
java -cp dist/lib/bigdl-0.2.0-SNAPSHOT-jar-with-dependencies-and-spark.jar \
com.intel.analytics.bigdl.models.vgg.Train \
--core core_number \
--node 1 \
--env local \
-f cifar10_folder \
-b batch_size
```
此命令使用以下参数：
- --core。训练中使用的计算机的物理核心数。
- --node。训练中使用的节点（计算机）数（在此示例中作为本地 Java 程序运行，因此设置为 `1`）。
- --env。可以是“local”或“spark”（在此情况下，设置为“local”）。
- -f。存放 CIFAR-10 数据集的文件夹。
- -b。最小批量大小（程序期望最小批量大小是 `node_number` × `core_number` 的倍数。在此示例中，`node_number` 为 `1`，您应将最小批量大小设置为 `core_number` × 4）。

运行 Spark 程序

您可以将 BigDL 程序，例如 VGG 模型训练，作为标准的 Spark 程序（在本地模式或集群模式下运行）来运行。要做到这一点，请执行以下步骤：

从 CIFAR-10 数据集页面下载 CIFAR-10 数据。
请记住选择二进制版本。
使用 `bigdl.sh` 脚本将示例作为 Spark 程序启动：
```
./dist/bin/bigdl.sh -- \
spark-submit --class com.intel.analytics.bigdl.models.vgg.Train \
dist/lib/bigdl-0.2.0-SNAPSHOT-jar-with-dependencies.jar \
--core core_number_per_node \
--node node_number \
--env spark \
-f cifar10_folder/ \
-b batch_size
```
此命令使用以下参数：
- --core。Spark 集群的每个执行器（或容器）中使用的物理核心数。
- --node。Spark 集群的执行器（容器）数量（在 Spark 本地模式下运行时，设置为 `1`）。
- --env。可以是“local”或“spark”（在此情况下，设置为“spark”）。
- -f。存放 CIFAR-10 数据集的文件夹（注意，在此示例中，这只是 Spark 驱动器上的本地文件文件夹。由于 CIFAR-10 数据集相对较小 [约 120 MB]，您会直接从驱动器发送到执行器）。
- -b。最小批量大小（期望最小批量大小是 `node_number` × `core_number_per_node` 的倍数。在此示例中，您应将最小批量大小设置为 `node_number` × `core_number_per_node` × `4`）。

调试常见问题

当前，BigDL 在模型训练中使用同步最小批量 SGD，并且在处理最小批量时，它会在每个执行器（或容器）上启动一个 Spark 任务（以多线程模式运行）。

将 `Engine.nodeNumber` 设置为 Spark 集群中的执行器数量。
确保每个 Spark 执行器具有相同数量的核心（`Engine.coreNumber`）。
最小批量大小必须是 `nodeNumber` × `coreNumber` 的倍数。

注意：在使用 Java 7 运行 BigDL for Spark 2.0 时，您可能会遇到性能不佳的情况，因此在为 Spark 2.0 构建和运行 BigDL 时请使用 Java 8。

在 Spark 2.0 中，由于 Kryo 问题 341：“由于 Generics 的 parentScope 被推送为同一对象而导致堆栈溢出错误”（有关更多信息，请参阅 Kryo 问题页面），请使用默认的 Java 序列化器而不是 Kryo。该问题已在 Kryo 4.0 中修复，但 Spark 2.0 使用的是 Kryo 3.0.3。Spark 版本 1.5 和 1.6 没有这个问题。

在 CentOS* 版本 6 和 7 上，将最大用户进程数增加到一个较大的值，例如 514585。否则，您可能会看到类似“无法创建新的本机线程”的错误。

当前，BigDL 在训练期间将所有训练和验证数据加载到内存中。如果 BigDL 内存不足，您可能会遇到错误。

BigDL 示例

注意：如果您使用的是虚拟机，您的性能将不如在原生操作系统安装中好，这主要是由于虚拟机的开销和分配给它的有限资源。这是预期的，并且与真实的 Spark 性能无关。

在 MNIST 上训练 LeNet

深度学习的“hello world”示例是在 MNIST 数据库上训练 LeNet（一个卷积神经网络）。

图 7 显示了克隆的 LeNet 示例。

图 7. 克隆的 LeNet 示例

BigDL 程序以导入 `com.intel.analytics.bigdl`._ 开始。然后，它初始化引擎，包括执行器节点数、每个执行器的物理核心数，以及它是在 Spark 上运行还是作为本地 Java 程序运行。

  val sc = Engine.init(param.nodeNumber, param.coreNumber, param.env == "spark").map(conf => {
    conf.setAppName("Train Lenet on MNIST")
      .set("spark.akka.frameSize", 64.toString)
      .set("spark.task.maxFailures", "1")
    new SparkContext(conf)
  })

如果程序在 Spark 上运行，`Engine.init()` 将返回一个带有正确配置的 `SparkConf`，然后您可以使用它来创建 `SparkContext`。否则，程序将作为本地 Java 程序运行，并且 `Engine.init()` 返回 `None`。

初始化后，通过调用 `LeNet5()` 开始创建 LeNet 模型，它创建 LeNet-5 卷积网络模型，如下所示：

val model = Sequential()
model.add(Reshape(Array(1, 28, 28)))
    .add(SpatialConvolution(1, 6, 5, 5))
    .add(Tanh())
    .add(SpatialMaxPooling(2, 2, 2, 2))
    .add(Tanh())
    .add(SpatialConvolution(6, 12, 5, 5))
    .add(SpatialMaxPooling(2, 2, 2, 2))
    .add(Reshape(Array(12 * 4 * 4)))
    .add(Linear(12 * 4 * 4, 100))
    .add(Tanh())
    .add(Linear(100, classNum))
    .add(LogSoftMax())

接下来，通过创建 DataSet.scala 命令（取决于是在 Spark 上运行还是本地运行，是分布式数据集还是本地数据集）来加载数据，然后应用一系列 Transformer.scala 命令（例如，`SampleToGreyImg`、`GreyImgNormalizer` 和 `GreyImgToBatch`）。

  val trainSet = (if (sc.isDefined) {
      DataSet.array(load(trainData, trainLabel), sc.get, param.nodeNumber)
    } else {
      DataSet.array(load(trainData, trainLabel))
    }) 
     -> SampleToGreyImg(28, 28) 
     -> GreyImgNormalizer(trainMean, trainStd) 
     -> GreyImgToBatch(param.batchSize)

之后，您通过指定数据集、模型和标准来创建优化器——根据是否在 Spark 上运行，是分布式还是本地的——标准根据输入和目标，计算给定损失函数的梯度。

val optimizer = Optimizer(
    model = model,
    dataset = trainSet,
    criterion = ClassNLLCriterion[Float]())

最后，在可选地指定优化器的验证数据和方法后，通过调用 `Optimizer.optimize()` 来训练模型。

optimizer
    .setValidation(
      trigger = Trigger.everyEpoch,
      dataset = validationSet,
      vMethods = Array(new Top1Accuracy))
    .setState(state)
    .setEndWhen(Trigger.maxEpoch(param.maxEpoch))
    .optimize()

以下命令从命令行执行 LeNet 示例：

$ dist/bin/bigdl.sh -- \
  java -cp dist/lib/bigdl-0.2.0-SNAPSHOT-jar-with-dependencies-and-spark.jar \
  com.intel.analytics.bigdl.models.lenet.Train \
  -f ~/data/mnist/ \
  --core 1 \
  --node 1 \
  --env local \
  --checkpoint

以下是在 VM 上运行的 LeNet Spark 本地模式的示例命令：

注意：在运行 Spark 本地模式之前，导出 `SPARK_HOME=your_spark_install_dir` 和 `PATH=$PATH:$SPARK_HOME/bin`。

$ ./dist/bin/bigdl.sh \
  -- spark-submit \
  --master local[1] \
  --driver-class-path dist/lib/bigdl-0.2.0-SNAPSHOT-jar-with-dependencies.jar \
  --class com.intel.analytics.bigdl.models.lenet.Train \
    dist/lib/bigdl-0.2.0-SNAPSHOT-jar-with-dependencies.jar \
  -f ~/data/mnist/ \
  --core 1 \
  --node 1 \
  --env spark \
  --checkpoint ~/model/model_lenet_spark

图像分类

注意：有关运行预训练模型的图像推理的示例，请参阅 BigDL README.md。

将图像下载到目录 `ILSVRC2012_img_val.tar`。这是预训练模型的图像 `tar` 文件。例如，将其放在 `~/data/imagenet` 中，然后解压缩。接下来，下载模型 `resnet-18.t7`，例如将其放在 `~/model/resnet-18.t7`。

从命令行运行以下命令（此示例适用于 VM，因此内存大小设置为 `1g`）：

$ ./dist/bin/bigdl.sh --spark-submit --master local[1] \
  --driver-memory 1g \
  --executor-memory 1g \
  --driver-class-path dist/lib/bigdl-0.2.0-SNAPSHOT-jar-with-dependencies.jar \
  --class com.intel.analytics.bigdl.example.imageclassification.ImagePredictor \  
  dist/lib/bigdl-0.2.0-SNAPSHOT-jar-with-dependencies-and-spark.jar \
  --modelPath ~/model/resnet-18.t7/resnet-18.t7 \
  --folder ~/data/imagenet/predict_100 \
  --modelType torch -c 1 -n 1 \
  --batchSize 4 \
  --isHdfs false

程序的输出是一个包含两列的表：`*.JPEG` 文件名和数字标签。要检查预测的准确性，请参阅 `imagenet1000_clsid.txt` 标签文件。（注意，文件标签是 0 基的，而示例的输出是 1 基的。）

图像文件的前几行如下：

{0: 'tench, Tinca tinca',
 1: 'goldfish, Carassius auratus',
 2: 'great white shark, white shark, man-eater, man-eating shark, Carcharodon carcharias',
 3: 'tiger shark, Galeocerdo cuvieri',
 4: 'hammerhead, hammerhead shark',
 ...

以下是示例输出的前几行：

[ILSVRC2012_val_00025162.JPEG,360]
[ILSVRC2012_val_00025102.JPEG,958]
[ILSVRC2012_val_00025113.JPEG,853]
[ILSVRC2012_val_00025153.JPEG,867]
[ILSVRC2012_val_00025132.JPEG,229]
[ILSVRC2012_val_00025133.JPEG,5]

图像文件 `ILSVRC2012_val_00025133.JPEG`（图 8）已正确识别为标签 `5`，这对应于 `imagenet1000_clsid.txt` 文件中的条目“4: ‘hammerhead, hammerhead shark’” 。

图 8. 图像文件 ILSVRC2012_val_00025133.JPEG

要从 IntelliJ IDE 中运行分类示例，您必须在 VM 中运行时以略微不同的方式设置参数，如图 9 所示。

图 9. IntelliJ 集成开发环境中的分类示例

由于图 9 中的 IntelliJ IDE 未显示完整的行，因此 VM 选项如下：

-Dspark.master=local[1] 
-Dspark.executor.cores=1 
-Dspark.total.executor.cores=1 
-Dspark.executor.memory=1g 
-Dspark.driver.memory=1g 
-Xmx1024m -Xms1024m

程序参数如下：

ImageClassifier Program arguments: 
  --modelPath ~/model/resnet-18.t7/resnet-18.t7 
  --folder ~/data/imagenet/predict_100 
  --modelType torch 
  -c 1 
  -n 1 
  --batchSize 1 
  --isHdfs false

神经网络支持的广度和深度

BigDL 目前支持以下预配置的神经网络模型：

自动编码器
Inception
LeNet
Resnet
Rnn
VGG

有关更多信息，请参阅 BigDL 模型页面。

此外，BigDL 支持超过 100 个标准的神经网络“构建块”，允许您配置自己的拓扑（图 10）。

图 10. BigDL 标准神经网络构建块

有关更多信息，请参阅 BigDL 神经网络页面。

此外，BigDL 还提供 LeNet、ImageNet 和 TextClassification 的端到端实现开箱即用示例（图 11）。

图 11. BigDL 示例

有关更多信息，请参阅 BigDL 示例页面。

缩写

HDFS	Hadoop 分布式文件系统
IDE	集成开发环境
MKL	数学核心库
RDD	容错分布式数据集
RNN	循环神经网络
SGD	随机梯度下降
VM	虚拟机
YARN	Yet Another Resource Negotiator（另一个资源协调器）