Spark MLlib Deep Learning Convolution Neural Network (深度学习-卷积神经网络)3.1，mllibconvolution

3、Spark MLlib Deep Learning Convolution Neural Network (深度学习-卷积神经网络)3.1

http://blog.csdn.net/sunbow0

Spark MLlib Deep Learning工具箱，是根据现有深度学习教程《UFLDL教程》中的算法，在SparkMLlib中的实现。具体Spark MLlib Deep Learning(深度学习)目录结构：

第一章Neural Net(NN)

1、源码

2、源码解析

3、实例

第二章Deep Belief Nets(DBNs)

1、源码

2、源码解析

3、实例

第三章Convolution Neural Network(CNN)

1、源码

2、源码解析

3、实例

第四章 Stacked Auto-Encoders(SAE)

第五章CAE

第三章Convolution Neural Network (卷积神经网络)

1 源码

目前SparkMLlib Deep Learning工具箱源码的github地址为：

https://github.com/sunbow1/SparkMLlibDeepLearn

1.1 CNN代码 

package CNN

import org.apache.spark._
import org.apache.spark.SparkContext._
import org.apache.spark.rdd.RDD
import org.apache.spark.Logging
import org.apache.spark.mllib.regression.LabeledPoint
import org.apache.spark.mllib.linalg._
import org.apache.spark.mllib.linalg.distributed.RowMatrix

import breeze.linalg.{
  Matrix => BM,
  CSCMatrix => BSM,
  DenseMatrix => BDM,
  Vector => BV,
  DenseVector => BDV,
  SparseVector => BSV,
  axpy => brzAxpy,
  svd => brzSvd,
  accumulate => Accumulate,
  rot90 => Rot90,
  sum => Bsum
}
import breeze.numerics.{
  exp => Bexp,
  tanh => Btanh
}

import scala.collection.mutable.ArrayBuffer
import java.util.Random
import scala.math._

/**
 * types：网络层类别
 * outputmaps：特征map数量
 * kernelsize：卷积核k大小
 * k: 卷积核
 * b: 偏置
 * dk: 卷积核的偏导
 * db: 偏置的偏导
 * scale: pooling大小
 */
case class CNNLayers(
  types: String,
  outputmaps: Double,
  kernelsize: Double,
  scale: Double,
  k: Array[Array[BDM[Double]]],
  b: Array[Double],
  dk: Array[Array[BDM[Double]]],
  db: Array[Double]) extends Serializable

/**
 * CNN(convolution neural network)卷积神经网络
 */

class CNN(
  private var mapsize: BDM[Double],
  private var types: Array[String],
  private var layer: Int,
  private var onum: Int,
  private var outputmaps: Array[Double],
  private var kernelsize: Array[Double],
  private var scale: Array[Double],
  private var alpha: Double,
  private var batchsize: Double,
  private var numepochs: Double) extends Serializable with Logging {
//        var mapsize = new BDM(1, 2, Array(28.0, 28.0))
//        var types = Array("i", "c", "s", "c", "s")
//        var layer = 5
//        var onum = 10  
//        var outputmaps = Array(0.0, 6.0, 0.0, 12.0, 0.0)
//        var kernelsize = Array(0.0, 5.0, 0.0, 5.0, 0.0)
//        var scale = Array(0.0, 0.0, 2.0, 0.0, 2.0)
//        var alpha = 1.0
//        var batchsize = 50.0
//        var numepochs = 1.0

  def this() = this(new BDM(1, 2, Array(28.0, 28.0)),
    Array("i", "c", "s", "c", "s"), 5, 10,
    Array(0.0, 6.0, 0.0, 12.0, 0.0),
    Array(0.0, 5.0, 0.0, 5.0, 0.0),
    Array(0.0, 0.0, 2.0, 0.0, 2.0),
    1.0, 50.0, 1.0)

  /** 设置输入层大小. Default: [28, 28]. */
  def setMapsize(mapsize: BDM[Double]): this.type = {
    this.mapsize = mapsize
    this
  }

  /** 设置网络层类别. Default: [1"i", "c", "s", "c", "s"]. */
  def setTypes(types: Array[String]): this.type = {
    this.types = types
    this
  }

  /** 设置网络层数. Default: 5. */
  def setLayer(layer: Int): this.type = {
    this.layer = layer
    this
  }

  /** 设置输出维度. Default: 10. */
  def setOnum(onum: Int): this.type = {
    this.onum = onum
    this
  }

  /** 设置特征map数量. Default: [0.0, 6.0, 0.0, 12.0, 0.0]. */
  def setOutputmaps(outputmaps: Array[Double]): this.type = {
    this.outputmaps = outputmaps
    this
  }

  /** 设置卷积核k大小. Default: [0.0, 5.0, 0.0, 5.0, 0.0]. */
  def setKernelsize(kernelsize: Array[Double]): this.type = {
    this.kernelsize = kernelsize
    this
  }

  /** 设置scale大小. Default: [0.0, 0.0, 2.0, 0.0, 2.0]. */
  def setScale(scale: Array[Double]): this.type = {
    this.scale = scale
    this
  }

  /** 设置学习因子. Default: 1. */
  def setAlpha(alpha: Double): this.type = {
    this.alpha = alpha
    this
  }

  /** 设置迭代大小. Default: 50. */
  def setBatchsize(batchsize: Double): this.type = {
    this.batchsize = batchsize
    this
  }

  /** 设置迭代次数. Default: 1. */
  def setNumepochs(numepochs: Double): this.type = {
    this.numepochs = numepochs
    this
  }

  /** 卷积神经网络层参数初始化. */
  def CnnSetup: (Array[CNNLayers], BDM[Double], BDM[Double], Double) = {
    var inputmaps1 = 1.0
    var mapsize1 = mapsize
    var confinit = ArrayBuffer[CNNLayers]()
    for (l <- 0 to layer - 1) { // layer
      val type1 = types(l)
      val outputmap1 = outputmaps(l)
      val kernelsize1 = kernelsize(l)
      val scale1 = scale(l)
      val layersconf = if (type1 == "s") { // 每一层参数初始化
        mapsize1 = mapsize1 / scale1
        val b1 = Array.fill(inputmaps1.toInt)(0.0)
        val ki = Array(Array(BDM.zeros[Double](1, 1)))
        new CNNLayers(type1, outputmap1, kernelsize1, scale1, ki, b1, ki, b1)
      } else if (type1 == "c") {
        mapsize1 = mapsize1 - kernelsize1 + 1.0
        val fan_out = outputmap1 * math.pow(kernelsize1, 2)
        val fan_in = inputmaps1 * math.pow(kernelsize1, 2)
        val ki = ArrayBuffer[Array[BDM[Double]]]()
        for (i <- 0 to inputmaps1.toInt - 1) { // input map
          val kj = ArrayBuffer[BDM[Double]]()
          for (j <- 0 to outputmap1.toInt - 1) { // output map          
            val kk = (BDM.rand[Double](kernelsize1.toInt, kernelsize1.toInt) - 0.5) * 2.0 * sqrt(6.0 / (fan_in + fan_out))
            kj += kk
          }
          ki += kj.toArray
        }
        val b1 = Array.fill(outputmap1.toInt)(0.0)
        inputmaps1 = outputmap1
        new CNNLayers(type1, outputmap1, kernelsize1, scale1, ki.toArray, b1, ki.toArray, b1)
      } else {
        val ki = Array(Array(BDM.zeros[Double](1, 1)))
        val b1 = Array(0.0)
        new CNNLayers(type1, outputmap1, kernelsize1, scale1, ki, b1, ki, b1)
      }
      confinit += layersconf
    }
    val fvnum = mapsize1(0, 0) * mapsize1(0, 1) * inputmaps1
    val ffb = BDM.zeros[Double](onum, 1)
    val ffW = (BDM.rand[Double](onum, fvnum.toInt) - 0.5) * 2.0 * sqrt(6.0 / (onum + fvnum))
    (confinit.toArray, ffb, ffW, alpha)
  }

  /**
   * 运行卷积神经网络算法.
   */
  def CNNtrain(train_d: RDD[(BDM[Double], BDM[Double])], opts: Array[Double]): CNNModel = {
    val sc = train_d.sparkContext
    var initStartTime = System.currentTimeMillis()
    var initEndTime = System.currentTimeMillis()
    // 参数初始化配置
    var (cnn_layers, cnn_ffb, cnn_ffW, cnn_alpha) = CnnSetup
    // 样本数据划分：训练数据、交叉检验数据
    val validation = opts(2)
    val splitW1 = Array(1.0 - validation, validation)
    val train_split1 = train_d.randomSplit(sp