Spark PCA降维模型提取人脸特征（scala+python）

引言

构建Spark MLlib PCA降维模型，加载人脸图片，并从图片中提取人脸特征。关于PCA原理详解可见本人的另一篇博客：数据降维——PCA（主成分分析）算法原理

1. 数据集准备

处理的数据来自户外脸部标注集（LFW，Labeled Faces in the Wild），这个数据集包含13000多张主要从互联网上获得的公众人物的面部图片，并都用人名进行了标注。

LFW数据很大，所以下载其中的一个子集就可以了，这个子集选择名字以A字母开头的人的面部图片。

奉上数据集：http://vis-www.cs.umass.edu/lfw/lfw-a.tgz

下载完解压缩，然后就开始Coding了。

2. 文件预处理

我们加载的是图片文件集，spark提供一个方法叫做wholeTextFiles，允许我们一次操作整个文件，不同于只能逐行处理的TextFile方法。wholeTextFiles将返回一个由key-value组成的RDD，key是文件路径，value是整个文件的内容。对于我们来说，只需要文件路径，因此我们需要从RDD抽取key，同时要注意的是，文件路径格式以“file:”开始，这个前缀是Spark用来区分从不同的文件系统读取文件的标识（例如file://是本地文件系统，hdfs://是hdfs，s3n://是Amason S3文件系统），所以我们需要先将“file:/”去掉。

    val conf=new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.wholeTextFiles("E:\\test\\lfw\\*")
    val files = rdd.map { case (fileName, content) => {
      fileName.replace("file:/", "")
    }}

3. 提取面部图片作为向量

每一个彩色图片可以表示成一个三维的像素数组或矩阵。前两维，即 $x$ ， $y$ 坐标，表示每个像素的位置，第三个维度表示每个像素的红、蓝、绿（RGB）三原色的值。

一个灰度图片每个像素仅仅需要一个值（不需要RGB值）来表示，因此可以简单表示为二维矩阵。很多和图片相关的图片处理和机器学习任务经常只处理灰度图片，我们将通过先把彩色图片转为灰度图片来达到这个目的。

在机器学习任务中，还有一种常用的方式是把图片表示成一个向量，而不是矩阵。通过连接矩阵的每一行（或者每一列）形成一个长向量（称为重塑）。这样每一个灰度图像矩阵会被转换为特征向量，作为机器学习模型的输入，下面使用的就是这种方式。

在这里的图像处理，主要是用了Java集成的AWT（抽象窗口工具库），分为下面三步，三步都封装为一个方法中：

（1）载入图片

首先读取人脸图片，这个不用多说啦。

  def loadImageFromFile(path:String):BufferedImage={
    ImageIO.read(new File(path))
  }

（2）转换灰度图片并改变图片尺寸

这二步的目的主要是为了提高处理效率，正如上面说的RGB彩色图片是三维矩阵，而灰度是二维的，那么使用RGB彩色图片而不是灰度图片会使得处理的数据增加三倍。类似地，较大的图片（较大指的是尺寸，高 $\times$ 宽）也大大增加了处理和存储的负担。

数据集中每个文件都为250 $\times$ 250的原始图片，那么单张图片就包含了187500个使用三原色的数据点。而如果我们转换为灰度图片，并改变图片尺寸，比方50 $\times$ 50像素大小，我们仅仅需要2500个数据点存储每个图片。这样对于处理整个数据集来说，就可以大大减少内存的消耗。下面将灰度转换和尺寸改变等实现都封装到processImage方法中：

  //处理图片：传入原始图片以及处理后图片的高宽像素(px)，将原始图片处理为灰度图片
  def processImage(image:BufferedImage,width:Int,height:Int):BufferedImage={
    val bwImage=new BufferedImage(width,height,BufferedImage.TYPE_BYTE_GRAY)
    val g = bwImage.getGraphics()
    g.drawImage(image,0,0,width,height,null)
    g.dispose()
    bwImage
  }

可以试一下效果，先加载一张图片，然后再调用processImage方法将其处理为150 $\times$ 150的灰度图片：

  def main(args: Array[String]): Unit = {
    val image = loadImageFromFile("E:\\test\\lfw\\Aaron_Eckhart\\Aaron_Eckhart_0001.jpg")
    val grayImage = processImage(image,150,150)
    ImageIO.write(grayImage,"jpg",new File("E:\\test\\image-output\\gray.jpg"))
  }

对比处理前的图片，左图为原图，右图为处理后的图片。

（3）提取特征向量

处理的最后一步是提取真实的特征向量作为我们降维模型的输入，正如上面提到的，我们这里要将纯灰度像素数据作为特征。通过打平二维的像素矩阵来构造一维向量，为此我们需要获取图片的全部像素点，可使用BufferedImage的getPixels()方法。

getPixels()方法用于提取图片中某个区域的像素点数据，因为我们要提取的是整张图片的像素点，所以总像素点pixels为图片的高 $\times$ 宽，即 $pixels=width\times height$ 。提取得到的所有像素点最后都封装在一个Array[Double]中，代码如下：

  //灰度图片传入，获取该图片的所有像素点
  def getPixelsFromImage(image:BufferedImage):Array[Double]={
    val width = image.getWidth
    val height = image.getHeight
    val pixels = Array.ofDim[Double](width*height)
    image.getData.getPixels(0,0,width,height,pixels)
  }

我们将上述三个方法都封装到方法extractPixels中，参数分别为：图片路径path，指定处理后的图片的宽width和高hieght。

  def extractPixels(path:String,width:Int,height:Int):Array[Double]={
    val raw: BufferedImage = loadImageFromFile(path)
    val processed=processImage(raw,width,height)
    getPixelsFromImage(processed)
  }

把这个函数应用到包含图片路径的RDD的每一个元素上将产生一个新的RDD，新的RDD包含每张图片的像素数据，让我们通过下面的main代码看看每一张图片的前10个像素点数据，也就是要将我们处理好的灰度图片转换为Double数组，并去该数组的前10个值：

  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.wholeTextFiles("E:\\test\\lfw\\*")
    val files = rdd.map { case (fileName, content) => {
      fileName.replace("file:/", "")
    }}
    val pixels: RDD[Array[Double]] = files.map(f=>extractPixels(f,50,50))
    println(pixels.take(10).map(_.take(10).mkString("",",",",...")).mkString("\n"))
  }

输出部分图片的前10个像素点数据如下：

至此，我们已经写好了处理图像获取像素点的部分，所以接下来就是为每一张图片都创建一个MLlib向量对象，并将其缓存下来加速我们之后的计算，在main中添加以下代码：

    val imageVectors = pixels.map(p=>Vectors.dense(p))
    imageVectors.setName("image-vectors").cache()

4. 标准化

在运行降维模型尤其是PCA之前，通常会对输入数据进行标准化（自行百度了解一下什么是标准化哈），Spark MLlib提供标准化的函数StandardScalar，该函数的两个参数分别为withMean、withStd，两个参数的值都是true or false。

withMean表示是否每个维度都减去他们的均值，只适用于稠密数据，不适用于稀疏数据，因为对于稀疏数据，提取平均值将会使之变稠密，往往不是我们想要的结果；withStd则表示是否除以标准差。在本例中，我们只从数据中提取平均值，所以withMean=true，withStd=false。我们对上述已缓存的图片向量进行标准化，并重新构建标准化后的向量，在main中添加如下代码：

    val scaler = new StandardScaler(withMean=true,withStd=false).fit(imageVectors)
    val scaledVectors = imageVectors.map(v=>scaler.transform(v))

5. 训练PCA降维模型

我们已经从图像的像素数据中获取到了向量，现在可以初始化一个新的RowMatrix，并调用computerPrincipalComponents来计算我们的分布式矩阵的前 $10$ 个主成分，在main中添加如下代码：

    val matrix = new RowMatrix(scaledVectors)
    val pc=matrix.computePrincipalComponents(10)

结果如何？我们可以打印出得到的主成分矩阵有几行几列：

    val rows = pc.numRows
    val cols = pc.numCols
    println(pc.numRows,pc.numCols)

从控制台可以看到结果，主成分矩阵有2500行10列：(2500,10)

解释：因为每种图片的维度是50 $\times$ 50，所以我们得到了前10个主成分向量，每一个向量的维度都和输入图片的维度一样，可以认为这些主成分是一组包含了原始数据主要变化的隐藏特征。在面部识别和图像处理中，这些主成分被称为特征脸。

6. 可视化特征脸

因为每一个主成分都和原始图像有相同纬度，每一个成分本身可以看做是一张图像，这使得我们下面要做的可视化特征脸成为可能，在这里使用Breeze线性函数库和Python的numpy及matplotlib的函数来可视化特征脸。

为此我们需要创建一个Breeze DenseMatrix，Breeze的linalg包中提供了实用的方法把矩阵写到CSV文件中，这里要导入breeze的相关类：import breeze.linalg.{DenseMatrix,csvwrite}，如果没有这个包的话，需要在pom文件上添加相关依赖配置。我们将使用它把主成分保存为CSV文件，在main中添加如下代码：

    val pcBreeze = new DenseMatrix(rows,cols,pc.toArray)
    csvwrite(new File("E:\\test\\pc.csv"),pcBreeze)

运行一下程序，可能有点久，但是最后会生成一个csv文件，用excel打开该文件可以看到有2500 $\times$ 10的数据，大概如下：

接下来我们要用IPython加载这个矩阵（也就是生成的这个CSV文件）并以图像的形式可视化主成分，打开windows cmd，输入ipython -pylab，这时打开了ipython之后并输入读取CSV文件的代码：

pcs=np.loadtxt("E:/test/pc.csv",delimiter=",")
print(pcs.shape)

可以输出了(2500,10)，就可以确认读取的矩阵和保存的矩阵维度相同：

这时我们需要一个方法来显示图片，因此先自定义如下一个函数：

 def plot_gallery(images,h,w,n_row=2,n_col=5):
     """Helper function to plot a gallery of portraits"""
     plt.figure(figsize=(1.8 * n_col,2.4 * n_row))
     plt.subplots_adjust(bottom=0, left=.01, right=.99, top=.90, hspace=.35)
     for i in range(n_row * n_col):
         plt.subplot(n_row, n_col, i+1)
         plt.imshow(images[:,i].reshape((h, w)),cmap=plt.cm.gray)
         plt.title("Eigenface %d" % (i+1), size=12)
         plt.xticks(())
         plt.yticks(())

定义完之后将使用这个方法绘制前10个特征脸：

plot_gallery(pcs,50,50)

结果：

每个主成分脸都是可以解释的，但是和聚类一样，并不总能直接精确地解释每个主成分代表的意义。不过从这些图片我们可以看出，有些图像好像选择了方向性的特征（例如图像6和9），有些集中表现了发型（例如图像4、5、7和10），而其他的似乎和面部特征更相关，比如眼睛、鼻子和嘴（例如图像1、7和9）

7. 完整代码

package sparkMllib.dimensionalityReduction
import java.awt.image.BufferedImage
import javax.imageio.ImageIO
import java.io.File
import breeze.linalg.{DenseMatrix,csvwrite}
import org.apache.spark.mllib.feature.StandardScaler
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object PcaImageProcess {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.wholeTextFiles("E:\\test\\lfw\\*")
    val files = rdd.map { case (fileName, content) => {
      fileName.replace("file:/", "")
    }}
    val pixels: RDD[Array[Double]] = files.map(f=>extractPixels(f,50,50))
    val imageVectors = pixels.map(p=>Vectors.dense(p))
    imageVectors.setName("image-vectors").cache()
    val scaler = new StandardScaler(withMean=true,withStd=false).fit(imageVectors)
    val scaledVectors = imageVectors.map(v=>scaler.transform(v))
    val matrix = new RowMatrix(scaledVectors)
    val pc=matrix.computePrincipalComponents(10)
    val rows = pc.numRows
    val cols = pc.numCols
    val pcBreeze = new DenseMatrix(rows,cols,pc.toArray)
    csvwrite(new File("E:\\test\\pc.csv"),pcBreeze)
  }

  //读取图片
  def loadImageFromFile(path:String):BufferedImage={
    ImageIO.read(new File(path))
  }

  //处理图片：传入原始图片以及处理后图片的高宽像素(px)，将原始图片处理为灰度图片
  def processImage(image:BufferedImage,width:Int,height:Int):BufferedImage={
    val bwImage=new BufferedImage(width,height,BufferedImage.TYPE_BYTE_GRAY)
    val g = bwImage.getGraphics()
    g.drawImage(image,0,0,width,height,null)
    g.dispose()
    bwImage
  }

  //灰度图片传入，获取该图片的所有像素点
  def getPixelsFromImage(image:BufferedImage):Array[Double]={
    val width = image.getWidth
    val height = image.getHeight
    val pixels = Array.ofDim[Double](width*height)
    image.getData.getPixels(0,0,width,height,pixels)
  }

  //封装以上三个方法，最终实现：处理path路径的图片并进行灰度，返回width*height的灰度图片的所有像素点
  def extractPixels(path:String,width:Int,height:Int):Array[Double]={
    val raw: BufferedImage = loadImageFromFile(path)
    val processed=processImage(raw,width,height)
    getPixelsFromImage(processed)
  }

}

引用及参考：

[1] 《Spark机器学习》 [南非]Nick Pentreath著

（欢迎转载，转载请注明出处）