345 lines
21 KiB
TeX
345 lines
21 KiB
TeX
%\documentclass[a4paper,12pt,UTF8,titlepage]{ctexart}
|
||
|
||
\documentclass[12pt,a4paper,titlepage]{article}
|
||
\usepackage{xltxtra,fontspec,xunicode}
|
||
\usepackage[slantfont,boldfont]{xeCJK}
|
||
|
||
\setmainfont{STSong} % 设置文档默认字体
|
||
|
||
\usepackage{setspace}%使用间距宏包
|
||
\usepackage{indentfirst}
|
||
\setlength{\parindent}{2em}
|
||
|
||
|
||
\usepackage{titlesec}
|
||
\titleformat*{\section}{\centering\huge\bfseries}
|
||
|
||
%页边距
|
||
\usepackage{geometry}
|
||
\geometry{left=2.0cm,right=2.0cm,top=2.5cm,bottom=2.5cm}
|
||
|
||
%页眉
|
||
\usepackage{fancyhdr}
|
||
\pagestyle{fancy}
|
||
\lhead{李志星 15060025 }
|
||
\chead{Spark MLlib应用报告}
|
||
\rhead{\leftmark}
|
||
|
||
%文档信息/同时也用于生成报告封面
|
||
\author{李志星\\ 15060025}
|
||
\date{2016年6月20日}
|
||
\title{\Huge 基于Spark MLlib的手写数字识别\newline \Large Logistic Regression的应用}
|
||
|
||
|
||
|
||
\usepackage{graphicx}
|
||
\usepackage{subfigure}
|
||
\DeclareGraphicsExtensions{.eps,.ps,.jpg,.bmp,.gif,.png}
|
||
|
||
\usepackage{pythonhighlight}
|
||
% "define" Scala
|
||
\usepackage{listings}
|
||
\lstdefinelanguage{scala}{
|
||
morekeywords={abstract,case,catch,class,def,%
|
||
do,else,extends,false,final,finally,%
|
||
for,if,implicit,import,match,mixin,%
|
||
new,null,object,override,package,%
|
||
private,protected,requires,return,sealed,%
|
||
super,this,throw,trait,true,try,%
|
||
type,val,var,while,with,yield},
|
||
otherkeywords={=>,<-,<\%,<:,>:,\#,@},
|
||
sensitive=true,
|
||
morecomment=[l]{//},
|
||
morecomment=[n]{},
|
||
morestring=[b]",
|
||
morestring=[b]',
|
||
morestring=[b]"""
|
||
}
|
||
\usepackage{color}
|
||
\definecolor{dkgreen}{rgb}{0,0.6,0}
|
||
\definecolor{gray}{rgb}{0.5,0.5,0.5}
|
||
\definecolor{mauve}{rgb}{0.58,0,0.82}
|
||
% Default settings for code listings
|
||
\lstset{frame=tb,
|
||
language=scala,
|
||
aboveskip=3mm,
|
||
belowskip=3mm,
|
||
showstringspaces=false,
|
||
columns=flexible,
|
||
basicstyle={\small\ttfamily},
|
||
numbers=none,
|
||
numberstyle=\tiny\color{gray},
|
||
keywordstyle=\color{blue},
|
||
commentstyle=\color{dkgreen},
|
||
stringstyle=\color{mauve},
|
||
frame=single,
|
||
breaklines=true,
|
||
breakatwhitespace=true
|
||
tabsize=3
|
||
}
|
||
|
||
\begin{document}
|
||
\begin{spacing}{1.5}
|
||
\maketitle
|
||
\section{实验内容}
|
||
|
||
\subsection{手写识别}
|
||
手写识别（Handwriting recognition）是计算机在纸、照片、触摸屏或其他设备中接收并识别人手写的文字等信息的技术，主要应用于光学字符识别（OCR）。手写识别系统能够用来识别汉字、英语、数字等字符。不过本报告的重点不在手写识别，而在于理解MLlib中的机器学习算法，因此以识别数字为例。识别数字0～9是个十类别问题，分类算法中最常用的场景是二分类，如果要用二分类解决这个问题，要采用one-against-one和one-against-all等做法进行处理，虽然复杂些，但是基本原理是一样的。因此本报告把主要关注点放在对MLlib算法的研究，只利用0和1的样本从而解决二分类问题。
|
||
|
||
\subsection{Logistic Regression}
|
||
在Zaharia提出Spark的论文中，他使用了Logistic Regression来作为一个实例来展示介绍Spark，因此在本报告中我也利用Logistic Regression来解决手写识别问题。Logistic Regression是比较常用的机器学习方法，用于估计某种事物的可能性，广告预测、用户行为预测和疾病预测等都经常用到它。它可以用来做回归，也可以用来分类，并且主要是二分类，在分类的时候，Logistic Regression不仅能分别出一个样本属于哪个类，还能计算出这个样本属于某个类类的可能性是多少。其实在本质上Logistic Regression是个线性回归，它只不过是在特征到结果的映射过程中多加了一个二值函数g映射：即先把样本的特征线性求和，接着把该连续值映射到0和1上。其大致的思路如下图所示：
|
||
\begin{figure}[h!]
|
||
\centering
|
||
\includegraphics[width=8cm]{LG_arch.png}
|
||
\caption{Logistic Regression原理示意图}
|
||
\label{LG_arch}
|
||
\end{figure}
|
||
|
||
这里w是模型参数，也就是回归系数，g是sigmoid函数，这个函数实际上是由对数几率变换的来的。在对Logistic Regression求解的时候，我见过最多的介绍的都是随机梯度下降（SGD）。梯度下降是利用一阶的梯度信息找到函数局部最优解的一种方法，也是机器学习里面最简单最常用的一种优化方法。它的思想很简单，要找最小值，只需要每一步都往下走（也就是每一步都可以让代价函数小一点），然后不断的走，那肯定能走到最小值的地方，需要更快的到达最小值就要每一步都找下坡最快的地方，而这个下坡最快的方向，就是梯度的负方向了。梯度下降算法在每次更新回归系数的时候都需要遍历整个数据集（计算整个数据集的回归误差），该方法对小数据集尚可。但当遇到有数十亿样本和成千上万的特征时，就有点力不从心了，它的计算复杂度太高。改进的方法是一次仅用一个样本点（的回归误差）来更新回归系数，这个方法叫就是随机梯度下降算法。
|
||
|
||
\subsection{MLlib算法主要机制}
|
||
在MLlib中有许多的概念，其中对理解其算法比较重要的有：DataFrame，Pipeline，Transfromer和Estimator。
|
||
|
||
\begin{itemize}
|
||
\item DataFrame：MLlib使用SaprkSQL中DataFrame来操作数据集，是最近才添加的API。在MLLib模块中有两个包都可以用来调用机器学习算法：mllib和ml。mllibRDD操作，而ml基于DataFrame，ml是官方推荐使用的。通过DataFrame可以操作各种各样的数据：文本、图像和结构化数据等。DataFrame以命名列的方式组织的分布式数据集 ，等同于关系型数据库中的一个表，和R/Python中的DataFrame类似，不过进行了很多的优化。
|
||
\item Pipeline：在MLlib中有一个很关键的概念：Pipeline。在利用解决机器学习问题时，经常要用对数据进行一系列的处理，MLlib用Pipeline来表示这样的工作流，在Pipeline中,包含一组以一定顺序执行的 PipelineStage( Transformer和Estimator)。
|
||
\item Transformer：Transformer是对特征转换和学习得到的模型的抽象，每一个Transfromer都要实现transform()方法，它把一个DataFrame处理后得到另一个DataFrame,一般来说新的DataFrame比原来的DataFrame要多一些列。
|
||
\item Estimator：Estimator是对一些机器学习算法或者其他的数据处理算法的抽象，每一个Estimator都有一个方法fit(),它以DataFrame为参数，返回一个模型，也就是Transformer，比如在MLlib中 LogisticRegression 就是一个Estimator，而LogisticRegressionModel就是一个Transformer。
|
||
\end{itemize}
|
||
|
||
一个比较简单形象的例子就是文本文档的处理，如下图*所示,上面一行表示了一个包含3个stage的Pipeline，前两个Tokenizer和HashingTF是Transformer,第三个LogisticRegression是一个Estimer。下面一行是一个流经pipeline的数据流,圆柱体代表了DataFrame.当调用pipeline的fit（）方法时， Tokenizer.transform() 把原始的Raw text切分成单词，然后把这些单词添加到原DataFrame中形成新的DataFrame从而让HashingTF.transfore()进行处理，他把单词列转化成特征矩阵又添加到新的DataFrame中，最后通过 LogisticRegression.fit()方法产生LogisticRegressionModel。生成的模型可以用于后续的测试等。
|
||
|
||
\begin{figure}[h!]
|
||
\centering
|
||
\includegraphics[width=16cm]{pipeline.jpg}
|
||
\caption{MLlib pipeline}
|
||
\label{pipeline}
|
||
\end{figure}
|
||
|
||
|
||
|
||
\section{实验过程}
|
||
\subsection{数据集}
|
||
本报告用到的数据集是MNIST的手写识别数据集，原始数据集分为多个文件，分别是训练数据和测试数据的特征值和类标号。这些文件存储的都是二进制格式，处理起来比较不方便，因此我在网上又找了一些经过别人处理过了的版本，它把数据转换成txt文件，一个文本文件对应一个样本，文件名表示了类标号和序号，文件里面包含一个32*32的0/1矩阵，矩阵中每一个点可以看成是手写图像中对应的一个像素点（如下图*）。训练数据和测试数据中分别有400个样本左右，虽然样本量有些小，但是对于体验一下Spark机器学习算法的还是可以的。
|
||
|
||
\begin{figure}[h!]
|
||
\centering
|
||
\subfigure[数字0对应的一个样本]{
|
||
\label{0} %% label for first subfigure
|
||
\includegraphics[width=1.0in]{data_0.jpg}}
|
||
\hspace{0.2in}
|
||
\subfigure[数字1对应的一个样本]{
|
||
\label{1} %% label for second subfigure
|
||
\includegraphics[width=1.0in]{data_1.jpg}}
|
||
|
||
\caption{样本示例}
|
||
\label{0_1} %% label for entire figure
|
||
\end{figure}
|
||
|
||
\subsection{代码}
|
||
如前言中所述，MLlib中有两个用于机器学习的包mllib和ml，根据应用趋势和其官方网站的建议，我采用了ml。Spark本身使用Scala语言编写的，但是它同样为Java、Python和R提供了几乎一致的接口，在本次报告中我使用的python。代码解释如下：
|
||
|
||
\subsubsection{导入依赖}
|
||
此段代码导入需要用到的包，包括数据处理的Vectors、算法训练的 LogisticRegression、算法评估BinaryClassificationEvaluator以及其他的一些用于和Spark操作的包。
|
||
\begin{python}
|
||
from pyspark import SparkContext
|
||
from pyspark.sql import SQLContext
|
||
from pyspark.ml.classification import LogisticRegression
|
||
from pyspark.mllib.linalg import Vectors
|
||
from os import listdir
|
||
from pyspark.ml.evaluation import BinaryClassificationEvaluator
|
||
\end{python}
|
||
|
||
\subsubsection{初始化环境}
|
||
SparkContext是在写Spark程序时入口，用来连接Spark并进行后续的操作，一般还会结合SparkConf对象来设置对Saprk集群的配置。这里我们用默认的设置即可。
|
||
SQLContext用来创建DataFrame。
|
||
\begin{python}
|
||
sc = SparkContext(appName="PythonlogExample")
|
||
sqlContext = SQLContext(sc)
|
||
\end{python}
|
||
|
||
\subsubsection{加载数据}
|
||
用于训练的DataFrame中应该包含两列：特征向量和类别。其中类别是数字1或者0.特征向量就是把32*32的矩阵转换成一个1024维的向量即可。load\_data函数接受一个表示训练样本的所在的目录的参数，遍历该目录下所有的文件也就是样本，从样本的名字解析出它的类别是0还是1，从文件内容中读取特征向量。然后生成DataFrame数据并返回。
|
||
|
||
\begin{python}
|
||
def load_data(data_folder):
|
||
file_list=listdir(data_folder)
|
||
file_num=len(file_list)
|
||
datas = list()
|
||
file_num=len(file_list)
|
||
datas = list()
|
||
for i in range(file_num):
|
||
filename=file_list[i]
|
||
fr=open('%s/%s' %(data_folder,filename))
|
||
data_in_line = list()
|
||
for j in range(32):
|
||
line_str=fr.readline()
|
||
for k in range(32):
|
||
data_in_line.append(int(line_str[k]))
|
||
label = filename.split('.')[0].split("_")[0]
|
||
datas.append((float(label),Vectors.dense(data_in_line)))
|
||
|
||
return sqlContext.createDataFrame(datas,["label","features"])
|
||
\end{python}
|
||
|
||
\subsubsection{模型训练}
|
||
在加载完训练数据后，即可用LogisticRegression来对其进行训练。新建LogisticRegression对象时可以指定一些参数，我在这里指定了最大迭代数和正则化参数。调用LogisticRegression的fit函数即可生成相应的LogisticRegressionModel。
|
||
|
||
\begin{python}
|
||
train_df = load_data("train")
|
||
lr = LogisticRegression(maxIter=10)
|
||
lrModel = lr.fit(train_df)
|
||
\end{python}
|
||
|
||
\subsubsection{模型评估}
|
||
利用测试数据对训练得到的模型进行评估，BinaryClassificationEvaluator用于评估二分类结果，我最后利用其提供的两个指标areaUnderROC和areaUnderPR中的areaUnderROC评价了一下该模型。areaUnderROC是ROC曲线右下角部分占正方形格子的面积比例，该值越大说明分类的效果越好。
|
||
\begin{python}
|
||
test_df = load_data("test")
|
||
predictions = lrModel.transform(test_df)
|
||
evaluator = BinaryClassificationEvaluator(labelCol="label", rawPredictionCol="rawPrediction", metricName="areaUnderPR")
|
||
accuracy=evaluator.evaluate(predictions)
|
||
print("Test Error = %g " % (1.0 - accuracy))
|
||
\end{python}
|
||
|
||
\section{实验结果}
|
||
在本次试验中，我分别设置最大迭代次数为0,5,10，来观察算法的训练效果。利用训练得到模型在测试数据上预测得到DataFrame predictions，然后调用predictions的show()方法，该方法会打印出前20个结果，一共五列，分别是label,features,rawPrediction,probability和prediction。label和features这两个列是加载的测试数据的类标号和特征向量。rawPrediction,probability和prediction是利用分别对训练数据训练以及模型对测试数据进行预测得到的，我查了一下相关文档，rawPrediction和probability都是一个向量，其维数就是类标号的个数，rawPrediction字面上的意思是原始的预测，直观的讲就是每个类的置信度（confidence），并且这个向量中的所有元素相加得到的和是0,probability是在给定rawPrediction的条件概率（该样本属于每个类的可能性），其计算方法根据所采用的分类算法而不同。在Logistic Regression中是这样计算的 1/(1+exp(-rawPrediction)。prediction是最后对样本的预测，它是对rawPrediction利用argmax（）函数得到的，也就是取对应rawPrediction最大的那个类。
|
||
|
||
\begin{itemize}
|
||
\item maxIter=0时，算法没有进行迭代，因为测试数据中有0和1的样本数是一样的，因此它会设置每个类标号的概率为0.5，算法把每个样本都预测为0，这样就有一半的测试样本是错误的，最后的areaUnderROC为0.5。\newline
|
||
\begin{figure}[h!]
|
||
\centering
|
||
\subfigure{
|
||
\label{0} %% label for first subfigure
|
||
\includegraphics[width=10cm]{show_0.jpg}}
|
||
\hspace{0.2in}
|
||
\subfigure{
|
||
\label{1} %% label for second subfigure
|
||
\includegraphics[width=8cm]{areaUnderROC_0.jpg}}
|
||
\caption{maxIter=0时预测值和areaUnderROC}
|
||
\label{0_1} %% label for entire figure
|
||
\end{figure}
|
||
|
||
|
||
\item maxIter=5时，areaUnderROC为0.9979。\newline
|
||
\begin{figure}[h!]
|
||
\centering
|
||
\subfigure{
|
||
\label{2} %% label for first subfigure
|
||
\includegraphics[width=10cm]{show_1.jpg}}
|
||
\hspace{0.2in}
|
||
\subfigure{
|
||
\label{3} %% label for second subfigure
|
||
\includegraphics[width=8cm]{areaUnderROC_1.jpg}}
|
||
\caption{maxIter=5时预测值和areaUnderROC}
|
||
\label{0_1} %% label for entire figure
|
||
\end{figure}
|
||
|
||
|
||
\item maxIter=10时，areaUnderROC为0.9993。\newline
|
||
\begin{figure}[h!]
|
||
\centering
|
||
\subfigure{
|
||
\label{4} %% label for first subfigure
|
||
\includegraphics[width=10cm]{show_2.jpg}}
|
||
\hspace{0.2in}
|
||
\subfigure{
|
||
\label{5} %% label for second subfigure
|
||
\includegraphics[width=8cm]{areaUnderROC_2.jpg}}
|
||
\caption{maxIter=10时预测值和areaUnderROC}
|
||
\label{0_1} %% label for entire figure
|
||
\end{figure}
|
||
|
||
\end{itemize}
|
||
|
||
从实验结果可以看出，随着迭代次数的增加，算法的效果也是越来越好。以rawPrediction或者probability为例，样本取对的类标签的置信度或者概率随着迭代次数的增加不断增加。
|
||
|
||
\section{库源代码分析}
|
||
如前所述，Spark 使用scala编写的，其Java接口是直接调用的scala实现，而Python接口又是基于Java接口实现的，因此虽然我报告中使用的Python接口实现的，但是算法的具体实现细节是用scala实现的，由于本人未曾使用过scala，因此在这里根据对代码的直观理解简要的介绍一下其大致思想。
|
||
|
||
\begin{itemize}
|
||
\item 参数设置，算法一开始都是一些基本的对参数的设置，比如迭代次数，规则化参数，是否对数据标准化等等。例如：
|
||
\begin{lstlisting}[escapeinside='']
|
||
@Since("1.2.0")
|
||
def setMaxIter(value: Int):this.type = set(maxIter,value)
|
||
setDefault(maxIter -> 100)
|
||
...
|
||
\end{lstlisting}
|
||
|
||
\item train()方法进行训练。首先是对训练数据集进行统计便于后续的处理。\newline
|
||
\begin{lstlisting}[escapeinside='']
|
||
val (summarizer, labelSummarizer) = {
|
||
//'此处用于生成相应的统计数据'
|
||
}
|
||
val histogram = labelSummarizer.histogram
|
||
val numInvalid = labelSummarizer.countInvalid
|
||
val numClasses = histogram.length
|
||
val numFeatures = summarizer.mean.size
|
||
\end{lstlisting}
|
||
|
||
其次是对训练数据集进行一些有效性和特殊性的判断。\newline
|
||
\begin{lstlisting}
|
||
if (numClasses > 2) {
|
||
...
|
||
}
|
||
\end{lstlisting}
|
||
由于在Spark MLlib中ml版本中LogisticRegression还未支持多分类，所以它会先判断样本是否是二分类，否则会先报异常。如果想用多分类的话可以用mllib版本。我看了下mlib版本的源码，它会进行判断，如果是二分类它会调用ml版本的实现，如果是多分类才会调用自己的实现。此外我发现Spark的帮助文档里有个小缺陷，在ml关于LogisticRegression的介绍中，他指出算法的详细实现细节可以看mllib的实现，但是mllib的实现中关于二分类并没有具体的算法细节，反而在ml的实现中是有的。
|
||
|
||
\begin{lstlisting}
|
||
} else if ($(fitIntercept) && numClasses == 2 && histogram(0) == 0.0) {
|
||
...
|
||
} else if ($(fitIntercept) && numClasses == 1) {
|
||
...
|
||
}
|
||
\end{lstlisting}
|
||
上面两个else分支是判断训练数据中的样本是否都是属于正样本否者负样本，与此同时，如果模型函数中还需要截距的话，那么其实不需要进行训练了，只要把系数都设为0，然后把截距设为正无穷或者负无穷即可。
|
||
|
||
|
||
\begin{lstlisting}
|
||
if (!$(fitIntercept) && numClasses == 2 && histogram(0) == 0.0) {
|
||
...
|
||
} else if (!$(fitIntercept) && numClasses == 1) {
|
||
...
|
||
}
|
||
\end{lstlisting}
|
||
上面两段代码也是判断训练数据中的样本是否都是属于正样本否者负样本，但是模型函数中如果不需要截距的话，那么本次训练可能就会出现问题，所以它会进行警告。
|
||
|
||
\begin{lstlisting}
|
||
val optimizer = if ($(elasticNetParam) == 0.0 || $(regParam) == 0.0) {
|
||
new BreezeLBFGS[BDV[Double]]($(maxIter), 10, $(tol))
|
||
} else {
|
||
...
|
||
new BreezeOWLQN[Int, BDV[Double]]($(maxIter), 10, regParamL1Fun, $(tol))
|
||
}
|
||
\end{lstlisting}
|
||
这段代码用来选择算法的优化器，在mllib的实现中，Logistic Regression有两种实现：LogisticRegressionWithSGD和LogisticRegressionWithLBFGS，并且官方推荐使用LogisticRegressionWithLBFGS。而在最新的ml的版本中，已经看不到SGD版本的了，他根据用户传参情况使用LBFGS或者OWLQN。
|
||
至此，算法的主要逻辑已经分析的差不多了，剩余的就是对模型函数的系数和截距的计算了，比如：
|
||
|
||
\begin{lstlisting}
|
||
val rawCoefficients = state.x.toArray.clone()
|
||
var i = 0
|
||
while (i < numFeatures) {
|
||
rawCoefficients(i) *= { if (featuresStd(i) != 0.0) 1.0 / featuresStd(i) else 0.0 }
|
||
i += 1
|
||
}
|
||
|
||
if ($(fitIntercept)) {
|
||
(Vectors.dense(rawCoefficients.dropRight(1)).compressed, rawCoefficients.last,
|
||
arrayBuilder.result())
|
||
} else {
|
||
(Vectors.dense(rawCoefficients).compressed, 0.0, arrayBuilder.result())
|
||
}
|
||
\end{lstlisting}
|
||
\end{itemize}
|
||
\newpage
|
||
\section{总结}
|
||
通过本次实验，我对Spark尤其是其MLlib库有了初步的了解。在如今大数据盛行的时代，Saprk有广泛的应用场景：流处理、交互式SQL查询和机器学习等。尤其是在机器学习方面，它基于内存的处理架构很适合机器学习的迭代运算。对于Spark来说，2015年最重要的变化应该是DataFrame的引入，它能够适应更广泛的数据处理要求，也吸引了更多的使用者。而随着DataFrame，起初的机器学习库mllib也逐渐迁移到基于DataFrame接口的库ml。在本次实验中，我感受到的有以下几点：
|
||
\begin{itemize}
|
||
\item Spark官方文档很完善，并且讲述的也很详细，对我这样的Spark初学者来说有很大的帮助。
|
||
\item Spark对API的支持很棒，虽然Spark使用scala开发的，但是它还提供了几乎一致的针对Java、python和R的接口，这能使更多语言背景的开发者轻松的使用Saprk进行开发。
|
||
\item 机器学习库工具比较完善，虽然目前Saprk MLlib实现的只是一些常用的算法，但是它却提供了很多的工具类，比如能够方便地对数据进行切分，拆分出训练数据和测试数据，对于训练的模型也提供了能够进行专业评估的工具类。因此 用Spark开发机器学习应用很方便，效率很快。
|
||
\end{itemize}
|
||
|
||
很感谢谭老师给我们布置这样的作业，让我接触到了Spark，为我以后的研究工作储备了更多的方法。否则我很难有机会主动地去接触这些新鲜的大数据技术，通过对Spark的了解熟悉，我也对软件工程中的软件设计和架构有了进一步的理解。总之，通过本次报告，我受益匪浅。
|
||
\end{spacing}
|
||
\end{document}
|