Spark上的分布式训练
随着复杂度的增加，深度学习模型可以获得高度的计算密集度。处理它的标准方法是使用更快的硬件（GPU）、代码优化或在分布式计算集群（如Spark）上训练网络。
建立一个Spark集群对数据科学家来说是一个难题，因为它需要大量的内存和集群配置。幸运的是，在SKIL实验中的Zeppelin笔记本提供了一个已经配置好的SparkContext，它可以被用于DL4J中分布式网络训练的Spark包装器所使用。
使用Spark集群
你将需要在后端安装一个Spark集群，并将分布式存储连接到该集群，比如HDFS。如果你已经完成了所有设置，你可以访问Spark解释器为你的笔记本配置它。
在这里，我们将重点介绍主要的工作流程和概念，以使分布式训练在skil内的Spark集群上工作。有关使用DL4J的ApacheSpark分布式训练的详细概述，请访问此处。
如果你已经熟悉了这些概念，你可以跳到以scala编写的Zeppelin笔记本的示例部分。
工作流程和概念
下面是分布式训练的主要概念和工作流程。
围绕网络配置的Spark包装器
与DL4J的多层网络MultiLayerNetwork和计算图ComputationGraph类类似，DL4J定义了两个用于在Spark上训练神经网络的类：
- SparkDl4jMultiLayer, MultiLayerNetwork的包装器
- SparkComputationGraph, ComputationGraph的包装器
因为这两个类是围绕标准单机器类的包装器，所以网络配置过程（即创建MultiLayerConfiguration或ComputationGraphConfiguration）在标准和分布式训练中是相同的。但是， Spark分布式与本地训练在两个方面有所不同：如何加载数据，以及如何设置训练（需要一些特定于集群的额外配置）。
TrainingMaster 类
DL4J中的TrainingMaster是一个抽象（接口），允许将多个不同的训练实现与SparkDl4jMultiLayer和SparkComputationGraph一起使用。
目前，DL4J有一个实现，即ParameterAveragingTrainingMaster。基本的TrainingMaster如下：
import org.deeplearning4j.spark.impl.paramavg.ParameterAveragingTrainingMaster
//spark训练配置: see http://deeplearning4j.org/spark for
//解释这些配置选项
//指定每个DataSet对象中有多少个示例
val tm = new ParameterAveragingTrainingMaster.Builder(dataSetObjectSize)
//平均和重新分布参数的频率
.averagingFrequency(5)
//如何处理异步预取多个小批量。0禁用预取，较大的值在预取时使用更多内存。
.workerPrefetchNumBatches(2)
//每个工作机线程的最小批处理大小：每个工作机线程中用于每个参数更新的示例数
.batchSizePerWorker(batchSize)
.build();
image.gif
可以在此处找到有关TrainingMaster Builder配置的更多信息。
分布式训练工作流程
在SKIL中的Spark集群上训练一个网络的典型工作流程如下:
- 从可用的SparkContext（sc）创建一个JavaSparkContext。
- 加载训练/测试数据并将其转换为RDD(
JavaRDD[DataSet])。 - 像通常使用MultiLayerNetwork和ComputationGraph类一样配置训练网络。
- 设置TrainingMaster ，以配置如何处理分布式训练。
- 包装你的网络配置 SparkDl4jMultiLayer 用于包装 **MultiLayerNetwork **或 SparkComputationGraph 用于包装 ComputationGraph.
- 调用
fit方法用于在你的数据上训练。 - 在测试数据上评估你的模型。
示例
1. 初始化JavaSparkContext
你可以从一个提供的SparkContext (sc)中创建一个JavaSparkContext
import org.apache.spark.api.java.JavaRDD
import org.apache.spark.api.java.JavaSparkContext
//从zeppelin的上下文sc创建spark context (sc)
val jsc = JavaSparkContext.fromSparkContext(sc)
image.gif
2. 加载数据并创建RDD
加载数据并创建JavaRDD[DataSet].
import org.deeplearning4j.datasets.iterator.impl.CifarDataSetIterator
import org.nd4j.linalg.dataset.DataSet
//将数据加载到内存中，然后并行化
//一般来说，这不是一个好方法-但对于本例来说，使用起来很简单
val batchSize = 32
val iterTrain = new CifarDataSetIterator(batchSize, 50000, true)
val iterTest = new CifarDataSetIterator(batchSize, 10000, false)
val trainDataList = new java.util.ArrayList[DataSet]
val testDataList = new java.util.ArrayList[DataSet]
while (iterTrain.hasNext()) {
trainDataList.add(iterTrain.next())
}
while (iterTest.hasNext()) {
testDataList.add(iterTest.next())
}
val trainData = jsc.parallelize(trainDataList)
val testData = jsc.parallelize(testDataList)
image.gif
3. 配置网络
网络配置将与标准设置相同。
import org.deeplearning4j.nn.conf.NeuralNetConfiguration
import org.deeplearning4j.nn.api.OptimizationAlgorithm
import org.nd4j.linalg.activations.Activation
import org.deeplearning4j.nn.weights.WeightInit
import org.nd4j.linalg.learning.config.Adam
import org.nd4j.linalg.lossfunctions.LossFunctions
import org.deeplearning4j.nn.conf.inputs.InputType
import org.deeplearning4j.nn.conf._
import org.deeplearning4j.nn.conf.layers._
//----------------------------------
//创建网络配置并进行网络训练
val conf = new NeuralNetConfiguration.Builder()
.seed(123)
.cacheMode(CacheMode.DEVICE)
.updater(new Adam(1e-2))
.biasUpdater(new Adam(1e-2*2))
//归一化以防止梯度消失或爆炸
.gradientNormalization(GradientNormalization.RenormalizeL2PerLayer)
.optimizationAlgo(OptimizationAlgorithm.STOCHASTIC_GRADIENT_DESCENT)
.l1(1e-4)
.l2(5 * 1e-4)
.list()
.layer(0, new ConvolutionLayer.Builder(Array[Int](4, 4), Array[Int](1, 1), Array[Int](0, 0)).name("cnn1").convolutionMode(ConvolutionMode.Same)
.nIn(3).nOut(64).weightInit(WeightInit.XAVIER_UNIFORM).activation(Activation.RELU)//.learningRateDecayPolicy(LearningRatePolicy.Step)
.biasInit(1e-2).build())
.layer(1, new ConvolutionLayer.Builder(Array[Int](4, 4), Array[Int](1, 1), Array[Int](0, 0)).name("cnn2").convolutionMode(ConvolutionMode.Same)
.nOut(64).weightInit(WeightInit.XAVIER_UNIFORM).activation(Activation.RELU)
.biasInit(1e-2).build())
.layer(2, new SubsamplingLayer.Builder(PoolingType.MAX, Array[Int](2, 2)).name("maxpool2").build())
.layer(3, new ConvolutionLayer.Builder(Array[Int](4, 4), Array[Int](1, 1), Array[Int](0, 0)).name("cnn3").convolutionMode(ConvolutionMode.Same)
.nOut(96).weightInit(WeightInit.XAVIER_UNIFORM).activation(Activation.RELU)
.biasInit(1e-2).build())
.layer(4, new ConvolutionLayer.Builder(Array[Int](4, 4), Array[Int](1, 1), Array[Int](0, 0)).name("cnn4").convolutionMode(ConvolutionMode.Same)
.nOut(96).weightInit(WeightInit.XAVIER_UNIFORM).activation(Activation.RELU)
.biasInit(1e-2).build())
.layer(5, new ConvolutionLayer.Builder(Array[Int](3, 3), Array[Int](1, 1), Array[Int](0, 0)).name("cnn5").convolutionMode(ConvolutionMode.Same)
.nOut(128).weightInit(WeightInit.XAVIER_UNIFORM).activation(Activation.RELU)
.biasInit(1e-2).build())
.layer(6, new ConvolutionLayer.Builder(Array[Int](3, 3), Array[Int](1, 1), Array[Int](0, 0)).name("cnn6").convolutionMode(ConvolutionMode.Same)
.nOut(128).weightInit(WeightInit.XAVIER_UNIFORM).activation(Activation.RELU)
.biasInit(1e-2).build())
.layer(7, new ConvolutionLayer.Builder(Array[Int](2, 2), Array[Int](1, 1), Array[Int](0, 0)).name("cnn7").convolutionMode(ConvolutionMode.Same)
.nOut(256).weightInit(WeightInit.XAVIER_UNIFORM).activation(Activation.RELU)
.biasInit(1e-2).build())
.layer(8, new ConvolutionLayer.Builder(Array[Int](2, 2), Array[Int](1, 1), Array[Int](0, 0)).name("cnn8").convolutionMode(ConvolutionMode.Same)
.nOut(256).weightInit(WeightInit.XAVIER_UNIFORM).activation(Activation.RELU)
.biasInit(1e-2).build())
.layer(9, new SubsamplingLayer.Builder(PoolingType.MAX, Array[Int](2, 2)).name("maxpool8").build())
.layer(10, new DenseLayer.Builder().name("ffn1").nOut(1024).updater(new Adam(1e-3)).biasInit(1e-3).biasUpdater(new Adam(1e-3*2)).build())
.layer(11,new DropoutLayer.Builder().name("dropout1").dropOut(0.2).build())
.layer(12, new DenseLayer.Builder().name("ffn2").nOut(1024).biasInit(1e-2).build())
.layer(13,new DropoutLayer.Builder().name("dropout2").dropOut(0.2).build())
.layer(14, new OutputLayer.Builder(LossFunctions.LossFunction.NEGATIVELOGLIKELIHOOD)
.name("output")
.nOut(10)
.activation(Activation.SOFTMAX)
.build())
.backprop(true)
.pretrain(false)
.setInputType(InputType.convolutional(32, 32, 3))
.build();
image.gif
4. 配置TrainingMaster
使用 ParameterAveragingTrainingMaster.Builder 来配置TrainingMaster
import org.deeplearning4j.spark.impl.paramavg.ParameterAveragingTrainingMaster
//Spark训练配置: 查看 http://deeplearning4j.org/spark for
//这些配置选项的解释
val tm = new ParameterAveragingTrainingMaster.Builder(batchSize)
.averagingFrequency(5)
.workerPrefetchNumBatches(2)
.batchSizePerWorker(batchSize)
.build();
image.gif
5. 包装为SparkDl4jMultiLayer
因为网络是一个MultiLayerNetwork配置，所以它被包装在SparkDl4jMultiLayer中。
import org.deeplearning4j.spark.impl.multilayer.SparkDl4jMultiLayer
//创建Spark网络
val sparkNet = new SparkDl4jMultiLayer(jsc, conf, tm)
image.gif
6. 训练网络
在训练数据上调用SparkDl4jMultiLayer#fit。
var x = 0
//执行训练:
for(x <- 0 to 50 ) {
sparkNet.fit(trainData)
println("Completed Epoch {}", x)
}
image.gif
7. 评估网络
在训练模型完成后在测试集上调用SparkDl4jMultiLayer#evaluate
//执行评估（分布式）
val evaluation = sparkNet.evaluate(testData)
println("***** Evaluation *****")
println(evaluation.stats())
//删除临时训练文件，现在已经完成了
tm.deleteTempFiles(jsc)
println("***** Example Complete *****")
image.gif
Keras模型配置的分布式训练
要在Keras模型配置上运行分布式训练，需要从Keras模型配置创建一个MultiLayerConfiguration或ComputationGraphConfiguration。查看此文档了解更多信息，特别是此部分。
在scala中，对于“MultiLayerConfiguration”，可以按以下方式进行：
import org.deeplearning4j.nn.modelimport.keras.KerasModelImport
val modelConfig = KerasModelImport.importKerasSequentialConfiguration("PATH TO YOUR JSON FILE")
image.gif
对于 "ComputationGraphConfiguration":
import org.deeplearning4j.nn.modelimport.keras.KerasModelImport
val computationGraphConfig = KerasModelImport.importKerasModelConfiguration("PATH TO YOUR JSON FILE")
image.gif
之后，可以按照这里的步骤操作，跳过步骤3。
下一步做什么?
Javadocs
以下是这里解释的重要类的javadocs：
