@R星校长

机器学习01【机器学习】

主要内容

1. Anconda安装使用
2. python开发spark
3. 线性回归算法原理
4. 线性回归算法案例

学习目标

第一节 pySpark环境准备

1).window配置python环境变量

window安装python，配置python环境变量。安装python后,在环境变量path中加入安装的路径，cmd中输入python，检验python是否安装成功。
注意：如果使用的是anaconda安装的python环境，建议安装python3.5.x版本，这个版本和spark1.6兼容。如何在anaconda中安装python3.5.x版本的python环境？参考文档：“Anaconda安装及使用.docx”

2).window Spark环境变量。
 window中配置Spark运行环境及Spark环境变量。
 a).解压spark的安装包到磁盘目录
 b).在环境变量中配置SPARK_HOME指定解压的路径,配置环境变量。
注意：解压spark安装包后，配置环境变量，一定配置到含有bin的外层。
3).python中安装py4j模块。
 在python环境中安装py4j模块(python调用java API的中间通信模块)
 进入C:\Python27\Scripts使用pip install py4j：
 pip install py4j 注：卸载python 模块：pip uninstall py4j
或者，将解压的spark安装包中的F:\spark-1.6.0-bin-hadoop2.6\python\lib\py4j-0.9-src\py4j拷贝C:\Python27\Lib\site-packages中。

验证py4j是否安装成功：进入python ,import py4j
注意：如果使用的是anaconda安装的python3.5.x的环境，之后使用这个python3.5.x环境，一定要将py4j模块放在安装的python3.5.x的目录，即：anaconda目录\envs\Python35【安装python3.5.x版本取的名称】\Lib\site-packages\中。

4).在python中安装pyspark模块

 使用pip安装pyspark：
  pip install pyspark ,会安装最新的版本的pyspark。
 或者，将解压的spark安装包中的F:\spark-1.6.0-bin-hadoop2.6\python\pyspark拷贝到C:\Python27\Lib\site-packages中，验证pyspark模块是否安装成功：

 进入cmd,输入python,导入pyspark模块，如果没错即安装成功。

注意：如果使用的是anaconda安装的python3.5.x的环境，之后使用这个python3.5.x环境，一定要将pyspark模块放在安装的python3.5.x的目录，即：anaconda目录\envs\Python35【安装python3.5.x版本取的名称】\Lib\site-packages\中。

第二节 pySpark代码开发

1. eclipse开发pySpark程序
 在eclipse中开发pySpark程序，需要安装pydev插件。

 1).eclipse安装python插件,安装完成后重启。
 2). 在window—>preferences中找到python interpreter配置安装python的路径：
 3).新建python项目：
2. pyCharm开发pySpark程序

 pyCharm是专为开发python的工具，在pyCharm中直接可以编写pySpark代码。

3. IDEA开发pySpark程序

 IDEA中开发pySpark代码需要安装python插件。

4. python wordcount

 pySpark WordCount 案例:

#coding:utf-8
#从pyspark中导入相应的包
from pyspark import SparkConf
from pyspark import SparkContext

def show(x):
    print x

if __name__ == '__main__':   
    conf = SparkConf()
    conf.setAppName("wordcount")
    conf.setMaster("local")
    sc = SparkContext(conf=conf)
    lines = sc.textFile("../../data/words", 2)
    print "lines rdd partition length = %d"%(lines.getNumPartitions())
    words = lines.flatMap(lambda line:line.split(" "), True)
    pairWords = words.map(lambda word : (word,1),True)
    result = pairWords.reduceByKey(lambda v1,v2:v1+v2, 3)
    print "result rdd partition length = %d"%(result.getNumPartitions())
    result.foreach(lambda t :show(t))
    #将结果保存到文件
    result.saveAsTextFile("../../data/wc-result")

eclipse console 控制台乱码问题：
eclipse控制台只支持GBK编码。运行时需要修改编码，运行python文件时，

右键->Run As->Run Configurations->Common->Encoding 改为GBK,没有GBK就直接输入运行。
5. python开发Spark原理

使用python api编写pyspark代码提交运行时，为了不破坏spark原有的运行架构，会将写好的代码首先在python解析器中运行(cpython)，Spark代码归根结底是运行在JVM中的，这里python借助Py4j实现Python和Java的交互,即通过Py4j将pyspark代码“解析”到JVM中去运行。例如，在pyspark代码中实例化一个SparkContext对象，那么通过py4j最终在JVM中会创建scala的SparkContext对象及后期对象的调用、在JVM中数据处理消息的日志会返回到python进程中、如果在代码中会回收大量结果数据到Driver端中，也会通过socket通信返回到python进程中。这样在python进程和JVM进程之间就有大量通信。
python开发spark，需要进行大量的进程间的通信，如果通信量过大，会出现“socket write error”错误，应尽量少使用回收数据类算子，也可以调节回收日志的级别，降低进程之间的通信。

6. python开发Spark问题

 1) 关于安装路径
 jdk，anaconda及python的安装路径中不能有空格和中文。
 2) 指定spark使用的python版本
 如果使用的anaconda更换了python3.5.x版本，之后在开发工具中指定了python解析器为3.5.x版本之后，运行python spark 代码时spark默认的使用的python版本使环境变量中指定的版本。会导致与指定的python解析器的python版本不一致。这时需要在环境变量中指定下PYSPARK_PYTHON环境变量即可，值为指定的python3.5.x python解析器路径。如：

第三节 pySpark案例

1. 统计PV,UV

1.	if __name__ == '__main__':
2.	    conf = SparkConf()
3.	    conf.setMaster("local")
4.	    conf.setAppName("test")
5.	    sc = SparkContext(conf=conf)
6.	
7.	    #pv
8.	    sc.textFile("./pvuv").map(lambda line:(line.split("\t")[4],1)).reduceByKey(lambda v1,v2:v1+v2).sortBy(lambda tp:tp[1],ascending=False).foreach(print)
9.	
10.	    #uv
11.	sc.textFile("./pvuv").map(lambda line:line.split("\t")[1]+"_"+line.split("\t")[4]).distinct().map(lambda one:(one.split("_")[1],1)).reduceByKey(lambda v1,v2:v1+v2).sortBy(lambda tp:tp[1],ascending=False).foreach(print)

2. 统计除了某个地区外的UV

1.	if __name__ == '__main__':
2.	    conf = SparkConf()
3.	    conf.setMaster("local")
4.	    conf.setAppName("test")
5.	    sc = SparkContext(conf=conf)
6.	
7.	    #uv
8.	    sc.textFile("./pvuv").filter(lambda line:line.split("\t")[3]=='beijing').map(lambda line:line.split("\t")[1]+"_"+line.split("\t")[4]).distinct().map(lambda one:(one.split("_")[1],1)).reduceByKey(lambda v1,v2:v1+v2).sortBy(lambda tp:tp[1],ascending=False).foreach(print)
9.	

3. 统计每个网站最活跃的top2地区

1.	def get_top2_local(one):
2.	    site = one[0]
3.	    local_iterable = one[1]
4.	
5.	    local_dic = {}
6.	    for local in local_iterable:
7.	      if local in local_dic:
8.	        local_dic[local] += 1
9.	      else:
10.	        local_dic[local] = 1
11.	
12.	    sorted_list = sorted(local_dic.items(),key = lambda x:x[1],reverse= True)
13.	    return_list = []
14.	    if(len(sorted_list)>=2):
15.	      for i in range(0,2):
16.	        return_list.append(sorted_list[i])
17.	      else:
18.	        return_list = sorted_list
19.	
20.	    return return_list
21.	
22.	
23.	if __name__ == '__main__':
24.	    conf = SparkConf()
25.	    conf.setMaster("local")
26.	    conf.setAppName("test")
27.	    sc = SparkContext(conf=conf)
28.	
29.	    #统计每个网站最活跃的top2地区
30.	    lines = sc.textFile("./pvuv")
31.	    site_local = lines.map(lambda line:(line.split("\t")[4],line.split("\t")[3]))
32.	    site_localIterable = site_local.groupByKey()
33.	    sorted_result = site_localIterable.map(lambda one:get_top2_local(one))
34.	    sorted_result.foreach(print)
35.	 

4. 统计每个网站最热门的操作

1.	def get_hot_operator(one):
2.	    site = one[0]
3.	    operator_iterable = one[1]
4.	
5.	    operator_dic = {}
6.	    for operator in operator_iterable:
7.	      if operator in operator_dic:
8.	        operator_dic[operator] += 1
9.	      else:
10.	        operator_dic[operator] = 1
11.	
12.	    sorted_list = sorted(operator_dic.items(),key = lambda x:x[1],reverse= True)
13.	    return_list = []
14.	    if(len(sorted_list)>=2):
15.	    for i in range(0,1):
16.	      return_list.append(sorted_list[i])
17.	    else:
18.	      return_list = sorted_list
19.	
20.	    return return_list
21.	
22.	
23.	if __name__ == '__main__':
24.	    conf = SparkConf()
25.	    conf.setMaster("local")
26.	    conf.setAppName("test")
27.	    sc = SparkContext(conf=conf)
28.	
29.	    #统计每个网站最热门的操作
30.	    lines = sc.textFile("./pvuv")
31.	    site_operator = lines.map(lambda line:(line.split("\t")[4],line.split("\t")[5]))
32.	    site_operatorIterable = site_operator.groupByKey()
33.	    sorted_result = site_operatorIterable.map(lambda one:get_hot_operator(one))
34.	    sorted_result.foreach(print)
35.	

5. 统计每个网站下最活跃的top3用户

1.	def get_uid_site_count(one):
2.	    uid = one[0]
3.	    site_iterable = one[1]
4.	
5.	    site_dic = {}
6.	    for site in site_iterable:
7.	      if site in site_dic:
8.	        site_dic[site] += 1
9.	      else:
10.	        site_dic[site] = 1
11.	
12.	    return_list = []
13.	    for site,count in site_dic.items():
14.	      return_list.append((site,(uid,count)))
15.	    return return_list
16.	
17.	def get_top3_uid(one):
18.	    site = one[0]
19.	    uid_count_iterable = one[1]
20.	    top3_uid = ['','','']
21.	    for tp in uid_count_iterable:
22.	      uid = tp[0]
23.	      count = tp[1]
24.	        for i in range(0,len(top3_uid)):
25.	          if(top3_uid[i]==''):
26.	            top3_uid[i] = tp
27.	            break
28.	          elif(count > top3_uid[i][1]):
29.	            for j in range(2,i,-1):
30.	              top3_uid[j] = top3_uid[j-1]
31.	              top3_uid[i] = tp
32.	              break
33.	
34.	       return top3_uid
35.	
36.	
37.	
38.	if __name__ == '__main__':
39.	    conf = SparkConf()
40.	    conf.setMaster("local")
41.	    conf.setAppName("test")
42.	    sc = SparkContext(conf=conf)
43.	
44.	    #统计每个网站最活跃的top3用户
45.	    lines = sc.textFile("./pvuv")
46.	    uid_site = lines.map(lambda line:(line.split("\t")[2],line.split("\t")[4]))
47.	    uid_siteIterable = uid_site.groupByKey()
48.	    uid_site_count = uid_siteIterable.flatMap(lambda one:get_uid_site_count(one))
49.	    top3_uid_info = uid_site_count.groupByKey().map(lambda one:get_top3_uid(one))
50.	    top3_uid_info.foreach(print)

第四节 线性回归算法

什么是回归？

从大量的函数结果和自变量反推回函数表达式的过程就是回归。线性回归是利用数理统计中回归分析来确定两种或两种以上变量间相互依赖的定量关系的一种统计分析方法。

一元线性回归：

多元线性回归：

如果回归分析中包括两个或两个以上的自变量，且因变量和自变量之间是线性关系，则称为多元线性回归分析。公式：

对于一元线性回归来说，如果现在有很多呈线性关系离散的点，假设现在需要找到一个线性回归公式来表示这些点自变量和因变量的关系，每两个点之间都能确定一条直线，如何找到一条直线来表示这些点的关系呢？

如何确定这条直线？

这个公式也叫做最小二乘法误差公式。error的值如果是0说明确定的这条直线穿过了所有的点，对于离散分散的一组点，error的值不可能为0。

确定权重的值（确定模型）
当误差公式:

如何确定error的最小值？


如何反向穷举求得线性回归的公式（训练线性回归模型）？

已知有线性回归关系的训练数据

已知线性回归的公式：
已知误差函数公式：

第五节 线性回归案例

1. 梯度下降法调节参数

判断模型error误差值收敛，也就是停止迭代的两种方式：

1. 指定一个error值，当迭代处理过程中得到的error值小于指定的值时，停止迭代。
2. 设置迭代次数，当迭代次数达到设置的次数时，停止迭代。

迭代周期：从调整参数开始到计算出error值判断是否大于用户指定的error是一个迭代周期。
当步长比较小时，迭代次数多，训练模型的时间比较长，当步长比较大时，有可能迭代的次数也比较多，训练模型的时间也会相对比较长，需要找到一个合适的步长。

2. 模型过拟合

训练模型都会将数据集分为两部分，一般会将0.8比例的数据集作为训练集，将0.2比例的数据集作为测试集，来训练模型。模型过拟合就是训练出来的模型在训练集上表现很好，但是在测试集上表现较差的一种现象，也就是模型对已有的训练集数据拟合的非常好（误差值等于0），对于测试集数据拟合的非常差，模型的泛化能力比较差。

如何判断模型发生过拟合？

训练出模型后，可以在训练集中测试下模型的正确率，在测试集中测试下模型的正确率，如果两者差别很大（测试集正确率小，训练集正确率大），那么模型就有可能发生了过拟合。

3. Spark Mllib 线性回归案例

object LinearRegression {

  def main(args: Array[String]) {
    // 构建Spark对象
    val conf = new SparkConf().setAppName("LinearRegressionWithSGD").setMaster("local")
    val sc = new SparkContext(conf)
    Logger.getRootLogger.setLevel(Level.WARN)
//        sc.setLogLevel("WARN")

    //读取样本数据
    val data_path1 = "lpsa.data"
    val data = sc.textFile(data_path1)
    val examples = data.map { line =>
      val parts = line.split(',')
      val y = parts(0)
      val xs = parts(1)
      LabeledPoint(parts(0).toDouble, Vectors.dense(parts(1).split(' ').map(_.toDouble)))
    }.cache()

    val train2TestData = examples.randomSplit(Array(0.8, 0.2), 1)

    /*
     *  迭代次数
     *  训练一个多元线性回归模型收敛（停止迭代）条件：
     *  	1、error值小于用户指定的error值
     *  	2、达到一定的迭代次数
     */
    val numIterations = 100

    //在每次迭代的过程中 梯度下降算法的下降步长大小    0.1 0.2 0.3 0.4
    val stepSize = 1

    
    val miniBatchFraction = 1
    val lrs = new LinearRegressionWithSGD()
    //让训练出来的模型有w0参数，就是由截距
    lrs.setIntercept(true)
    //设置步长
    lrs.optimizer.setStepSize(stepSize)
    //设置迭代次数
    lrs.optimizer.setNumIterations(numIterations)
    //每一次下山后，是否计算所有样本的误差值,1代表所有样本,默认就是1.0
    lrs.optimizer.setMiniBatchFraction(miniBatchFraction)

    val model = lrs.run(train2TestData(0))
    println(model.weights)
    println(model.intercept)

    // 对样本进行测试
    val prediction = model.predict(train2TestData(1).map(_.features))
    val predictionAndLabel = prediction.zip(train2TestData(1).map(_.label))
    
    val print_predict = predictionAndLabel.take(20)
    println("prediction" + "\t" + "label")
    for (i <- 0 to print_predict.length - 1) {
      println(print_predict(i)._1 + "\t" + print_predict(i)._2)
    }
    
    // 计算测试集平均误差
    val loss = predictionAndLabel.map {
      case (p, v) =>
        val err = p - v
        Math.abs(err)
    }.reduce(_ + _)
    val error = loss / train2TestData(1).count
    println(s"Test RMSE = " + error)
    // 模型保存
    val ModelPath = "model"
    model.save(sc, ModelPath)
//    val sameModel = LinearRegressionModel.load(sc, ModelPath)
    sc.stop()
  }

}

本节作业

1. 完成 pyspark 环境配置。
2. pySpark 代码开发。
3. 线性回归原理。