spark 学习笔记

朴灿烈づ我的快乐病毒、 2022-07-13 08:11 192阅读 0赞

# spark 学习笔记 #

《spakr快速大数据分析》  
书中例子是以spark1.2为基础写的。

# 第二章 #

启动 ipython `PYSPARK_DRIVER_PYTHON=ipython ./bin/pyspark`  
使用ipython notebook `PYSPARK_DRIVER_PYTHON=ipython PYSPARK_DRIVER_PYTHON_OPTS="notebook --pylab inline" ./bin/pyspark`  
在windows上启动 `set IPYTHON=1 \n bin\pyspark`

将py程序提交到spark执行 `bin/spark-submit my_script.py`

## 初始化sparkcontext ##

也就是创建到spark的连接

from pyspark import SparkConf, SparkContxt
    conf = SparkConf().setMaster("local").setAppName("my app")
    sc = SparkContext(conf=conf)
    
    # 关闭连接
    sc.stop()

# 第三章RDD编程 #

## 创建RDD ##

# 从文件读取数据
    line = sc.textFile("README.md")
    # parallelize 方法
    line = sc.parallelize(['pandas','i like pandas'])

## RDD转换操作 ##

inputRDD = sc.textFile('log.txt')
    errRDD = inputRDD.filter(lambda x:'error' in x)
    warnRDD = inputRDD.filter(lambda x:'warning' in x)
    bindRDD = errRDD.union(warnRDD)

## RDD行动操作 ##

bindRDD.count()
    bindRDD.take(10)
    # 返回全部数据集
    bindRDD.collect()

## 向spark传递函数 ##

# lambda 函数
    word = rdd.filter(lambda s:'python' in s)
    # def 定义的函数
    def containsErr(s):
        return 'error' in s
    word = rdd.filter(containsErr)

## RDD常见转换操作 ##

`以 rdd={1,2,3,3} 为例的转换操作`

map()
    # 将函数应用与RDD中的每个元素，将返回值构建新的RDD
    rdd.map(x => x+1)
    
    flatMap()
    # 将函数应用用RDD中的每个元素，将返回的迭代器的所有内容构成新的RDD。通常用于切分单词。
    rdd.flatMap(x=>x.to(3))  --> {1,2,3,2,3,3,3})
    
    filter()
    # 返回一个由通过传给filter()的函数的元素组成的RDD
    rdd.filter(x=>x!=1)  --> {2,3,4}
    
    distinct()
    # 去重
    rdd.distinct()  --> {1,2,3}
    
    sample(withReplacement,fraction,[seed])
    # 对RDD进行采样，以及是否替换
    rdd.sample(false,0.5)   --> 非确定的

`以{1,2,3}和{3,4,5}的RDD转换操作`

union()
    # 求并集
    rdd.union(other)  --> {
       1,2,3,4,5}
    
    intersection()
    # 求交集
    rdd.intersection(other)  --> {
       3}
    
    subtract()
    # 移除一个RDD中的内容，相当于减去一个交集
    rdd.subtract(other)  -->  {
       1,2}
    
    cartesian()
    # 与另一个RDD的笛卡尔积
    rdd.cartesian(other)  --> {(1,3),(1,4)...(3,5)}

## RDD常见行动操作 ##

`以{1,2,3,3}为列说明常见行动操作`

collect()
    # 返回RDD中所有的元素
    rdd.collect()  --> {
       1,2,3,3}
    
    count()
    # 计数
    rdd.count()
    
    countByValue()
    # 各元素在RDD中出现的次数
    rdd.countByValue()   --> {(1,1),(2,1),(3,2)}
    
    take(num)
    # 返回前n元素
    
    top(n)
    # 排序后的前n个元素
    
    takeOrdered(num)(ordering)
    # 按照指定顺序，从rdd中返回前n个元素
    rdd.takeOrdered(2)(myOrdering)   --> {
       3,3}
    
    takeSample(withReplacement,num,[seed])
    # 从RDD中返回任意一些元素
    rdd.takeSample(false,1)  --> 非确定的
    
    reduce()
    # 并行整合rdd中所有的数据，比如sum
    rdd.reduce((x,y)=>x+y) --> 9
    
    fold(zero)(func)
    # 和reduce()一样，但是需要提供初始值
    rdd.fold(0)((x,y)=>x+y) --> 9
    
    aggregate(zeroValue)(seqOp,combOp)
    # 和reduce类似，但是通常返回不同类型的函数
    aggregate((0,0))((x,y)=>(x._1+y,x._2+1), (x,y)=>(x._1+y._1,x._2+y._2) ) --> 9
    
    foreach(func)
    # 对RDD中的每个元素使用给定的函数
    rdd.foreach(func)

## 持久化缓存 ##

from pyspark.storage import StorageLvel
    rdd.presist(StoragLevel.DISK_ONLY)
    
    RDD.cache()
    
    # 缓存的级别
    # MEMORY_ONLY
    # MEMORY_ONLY_SER
    # MEMORY_AND_DISK # 如果内存放不下，则溢出写到磁盘上
    # MEMORY_AND_DISK_SER # 如果内存放不下，则溢出写到磁盘上，在内存中存放序列化后的数据
    # DISK_ONLY
    
    # 移除缓存
    RDD.unpersist()

# 第四章 键值对操作 #

## 键值对的RDD转化操作 ##

# 以{(1,2),(3,4),(3,6)}为例
    
    reduceByKey(func)
    # 合并具有形同键的值
    rdd.reduceByKey((x,y)=>x+y)  -->{(1,2),(3,10)}
    
    groupByKey()
    # 对具有相同键的值分组
    rdd.groupByKey()  --->{(1,[2]),(3,[4,6])}
    
    combineByKey(createCombiner,mergeValue,mergeComBiners,partitioner)
    # 使用不同的返回类型合并具有相同键的值。有多个参数分别对应聚合操作的各个阶段，因而非常适合用来解释聚合操作各个阶段的功能划分。
    # 下面是求每个键的平均值
    sumCount=num.combineByKey((lambda x:(x,1)),
                               (lambda x,y:(x[0]+y,x[1]+1)),
                               (lambda x,y:(x[0+y[0],x[1]+y[1]])))
    sumCount.map(lambda key,xy:(key,xy[0]/xy[1])).collectAaMap()
    
    mapValues(func)
    # 对pairRDD的每个值应用一个函数而不改变键
    rdd.mapVlues(x=>x+1)
    
    flatMapValues(func)
    # 对pairRDD的每个值应用一个返回迭代器的函数，然后对返回的每个元素都生成一个对应原键的键值对记录，通常用于符号化
    rdd.flatMapValues(x=>(x to 5))  -->{(1,2),(1,3),(1,4),(1,5),(3,4),(3,5)}
    
    keys()
    # 返回一个仅含有键的RDD
    rdd.keys()   ->{
       1,3,3}
    
    values()
    # 返回一个仅包含值的RDD
    rdd.values()   -->{
       2,4,6}
    
    sortByKey()
    # 返回一个根据键排序的RDD
    rdd.sortByKey(ascending=True)  -->{(1,2),(3,4),(3,6)}

## 两个键值对RDD的转换操作 ##

# 以rdd={(1,2),(3,4),(3,6)} other={(3,9)} 为例
    
    subtracByKey()
    # 删除rdd中键与other中键相同的元素
    rdd.subtracByKey(other)  --> {(1,2)}
    
    join()
    # 对两个rdd内链接
    rdd.join(other)   --> {(3,(4,9)),(3,(6,9))}
    
    rightOuterJoin()
    # 对两个rdd进行连接操作，确保第一个rdd中的键必须存在（右外链接）
    rdd.rightOuterJoin(other)   --> {(3,(some(4),9)),(3,(some(6),9))}
    
    leftOuterJoin()
    # 对两个rdd进行连接操作，确保第二个rdd中的键必须存在（左外连接）
    rdd.leftOuterJoin(other)  --> {(1,(2,None)),(3,(4,some(9))),(3,(6,some(9)))}
    
    congroup()
    # 将两个rdd中拥有相同键的数据分组到一起
    rdd.congroup(other)  --> {(1,([2],[])),(3,([4,6],[9]))}

## 并行度调优 ##

每个rdd都有固定数目的分区，分区数决定了在rdd上执行操作的并行度。  
大多数操作符都能接受第二个参数，用来指定分组结果或者聚合结果的rdd的分区数。  
`比如 sc.parallelize(data).reduceByKey(lambda x,y:x+y,10) 指定分区数10`  
查看分区数  
`rdd.partitions.size`或`rdd.getNumPartitions`  
改变分区的方法

repartition()
    
    # scala方式
    var rdd= sc.sequenceFile(.....).partitionBy(new spark.HashPartitioner(100))  # 构造100个分区
    rdd.partitoner
    
    # python方式
    rdd。partitionBy(100)

## 键值对Pair RDD的行动操作 ##

# 以 rdd={(1,2),(3,4),(3,6)} 为例
    
    countByKey()
    # 对每个键对应的元素分别计数
    rdd.countByKey()  --> {(1,1,),(3,2)})
    
    collectAsMap()
    # 将结果以映射表的形式返回，以便查询
    rdd.collectAsMap()  --> Map{(1,2),(2,6)}
    
    lookup(key)
    # 返回给定键对应的所有值
    rdd.lookup(3)   --> [4,6]

# 第五章 数据读取与保存 #

## 读取txt文件 ##

输入的每一行都会成为RDD的一个元素。

# 读取文件
    input=sc.textFile("file:///home/holden/README.md")
    # 保存文件
    result.saveAsTextFile(outputFile)

## 读取json ##

# 将json文件的每一行假设为一条记录来处理
    import json
    data = input.map(lambda x:json.load(x))
    # 写
    (data.filter(lambda x:x[lovesPandas"]).map(lambda x:json.dumps(x)).saveAsTextFile(outputFile))

## 读取csv ##

同样是将读取的文本的每一行当做一条记录

import csv
    from io import StringIO
    def loadRecord(line):
        """解析一行csv记录"""
        input = StringIO(line)
        reader = csv.DictReader(input,filednames=["name","favouriteAnimal"])
        return reader.next()
    input = sc.textFile(inputFile).map(loadRecord)
    
    # 保存csv
    def writeRecords(records):
        """写出一些csv记录"""
        output = StringIO()
        writer = csv.DictWriter(output,fieldnames=["name","favoriteAnimal"])
        for record in records:
            writer.writerow(record)
        return [output.getvalue()]
    pandaLovers.mapPartitions(writeRecords).saveAsTextFile(outputFile)

## 读取SequenceFile ##

## Hadoop输入输出格式 ##

## 关系型数据库 ##

## HBase ##

# 第六章 Spark进阶编程 #

## 两种类型的共享变量 ##

累加器（qccumulator）：用于对信息聚合，提供了将工作节点中的值聚合到驱动器程序中的简单语法。  
广播变量（broadcast variable）：用来高效分发较大的对象，让程序高效地向所有工作节点发送一个较大的值，以供一个或多个spark操作使用。

# 在python中累加空行,使用了累加器
    file = sc.textFile(inputFile)
    # 创建累加器并初始化为0
    blankLine=sc.accumulator(0)
    
    def extractCallSigs(line):
        global blankLine  # 访问全局变量
        if (line==""):
            blankLine+=1
        return line.split(" ")
    
    callSigns = file.flatMap(extractCallSigns)
    callSigns.saveAsTextFile(output)

# 使用广播变量查询国家
    # 查询rdd中呼叫号对应的位置，将呼号前缀读取为国家代码来进行查询
    signPrefixes = sc.broadcast(loadCallSignTable())   # 广播变量
    
    def processSignCount(sign_count,signPrefixes):
        country=lookupCountry(sign_count[0],signPrefixes.value)
        count = sign_count[1]
        return (country,count)
    
    countryContactCounts=(contactCounts.map(processSignCount).reduceByKey((lambda x,y:x+y)))
    countryContactCounts.saveAsTextFile(output)

广播变量的优化，选择合理的序列化库，可以加快对应的序列化和反序列化时间。

基于分区进行操作  
spark提供基于分区的map和foreach，使部分代码只对rdd的每个分区运行一次，可以帮助降低这些操作的代价。

# 按照分区执行的操作符
    
    mapPartitions()
    # 参数：该分区中元素的迭代器。返回：元素的迭代器
    # 对于RRD[T]的函数签名 ：f:(iterator[T]) --> iterator[U]
    
    mapPartitionsWithIndex()
    # 参数：分区序号，以及每个分区中的元素的迭代器。返回：元素的迭代器
    # 对于RRD[T]的函数签名 ：f:(int,iterator[T]) --> iterator[U]
    
    foreachPartitions()
    # 参数：元素迭代器。返回：无
    # 对于RRD[T]的函数签名 ：f:(iterator(T)) -->Unit

数值RDD的操作

count()
    # RDD中元素个数
    mean()
    # 元素平均值
    sum()
    # 
    max()
    min()
    variance()  # 方差
    sampleVariance()  # 从采样中计算出的方差
    stdev()  # 标准差
    sampleStdev()   # 采用的标准差

# 用python移除异常值
    要把string类型的rdd转成数值型，这样才能使用统计函数并移除异常值
    distanceNumerics=distances.map(lambda string:float(string))
    stats = distanceNumerics.stats()
    seddev=stats.sedev()
    mean =stats.mean()
    reasonableDistances=distanceNumerics.filter(lambda x:math.fabs(x-mean)<3*stddev)
    print reasonableDistances.collect()

# 第11章 基于MLlib的机器学习 #

## 逻辑回归的例子 ##

# 逻辑回归的垃圾邮件分类
    from pyspark.mllib.regression import LabeldPoint
    from pyspark.mllib.feature import HashingTF
    from pyspark.mllib.classification import LogisticRegressionWithSGD
    
    spam=sc.textFile('spam.txt')
    normal = sc.textFile('normal.txt')
    
    # 创建一个HashingTF实例来把邮件文本映射为包含10000个特征的向量
    tf=HashingTF(numFeatures=10000)
    # 各邮件都切分为单词，每个单词映射为一个特征
    spamFeatures = spam.map(lambda email: tf.transForm(email.split(' ')))
    normalFeatures = normal.map(lambda email: tf.transform(email.split(' ')))
    
    # 创建LabelPoint数据集分别存放阳性（垃圾邮件）和阴性（正常邮件）的例子
    positiveExample = spamFeatures.map(lambda features:LabeldPoint(1,features))
    negativeExamples = normalFeatures.map(lambda features:labeldPoint(0,features))
    trainingData = positiveExample.union(negativeExample)
    trainingData.cache()  # 因为逻辑回归是迭代算法，所以需要缓存训练数据RDD
    
    # 使用SGD算法
    model = LogisticRegressionWithSGD.train(trainningData)
    
    # 以阳性和阴性的例子分别测试。
    # 首先用一样的HashingTF特征来得到特征向量，然后对该向量应用得到的模型
    posTest = tf.transform("O M G GET cheap stuff by sending money to ...".split(' '))
    negTest = tf.transform("Hi Dad, i started studying spark the other ...".split(' '))
    print( "predict for postive test example:%g" % model.predict(posTest))
    print( "predict for negative test example:%g" % model.predict(negTest))

## 数据类型 ##

MLlib包含一些特有的数据类型，对于scala和Java，它们位于`org.apache.spark.mllib`下，对于python则是位于`pyspark.mllib`下。

Vector
    # 一个数学向量，既支持稠密向量，也支持稀疏向量
    # 创建方法　mllib.linalg.Vectors 类
    
    LabeldPoint
    # 用来表示带标签的数据点，它包含一个特征向量和一个标签(浮点数)
    #　位于mllib.regression包中
    
    Rating
    # 用户对一个产品的评分，用于产品推荐
    # 位于 mllib.recommendattion 包
    
    各种model
    # 每个model都是训练算法的结果，一般有一个predict方法可以用来对新的数据点或数据点组成的RDD应用改model进行预测

## 操作向量 ##

在Java和scala中，MLlib的Vector类主要用来为数据表示服务的，而没有在用户API中提供加法减法这样的向量运行。在scala中可以使用Breeze，在python中，可以对稠密向量使用numpy来进行数学操作，也可以把这些操作传给MLlib。

from numpy import array
    from pyspark.mllib.lanalg import Vectors
    
    # 创建稠密向量
    denseVec1 = array([1.0,2.0,3.0]})  # numpy数组可以直接传给mllib
    denseVec2 = Vectors.dense([1.0,2.0,3.0])  # 或者使用Vectors类来创建
    
    # 创建稀疏向量
    # 向量的维度(4)以及非零位的位置和对应的值
    # 这些数据可以用一个dictionary来传递，或者使用两个分别代表位置和值的list
    sparseVec1 = Vectors.sparse(4,{
       0:1,2:2.0})
    sparseVec2 = Vectors.sparse(4,[0,2],[1.0,2.0])

## 特征提取 ##

mllib.feature包中包含一些用来进行常见特征转换的类，包括从文本常见特征向量，对特征向量进行正则化和伸缩变换的方法。

### TF-IDF ###

MLlib中有两个算法可以用来计算TF-IDF：HashingTF 和 IDF 。

# python中使用 HashingTF
    
    from pyspark.mllib.feature import HashingTF
    
    sentence = "hello hello world"
    words =sentence.split()  # 将句子切分为一串单词
    tf = HashingTF(10000)  # 将所有单词映射到[0,S-1]之间的数字上，这里S=10000
    tf.transform(words)
    
    rdd = sc.wholeTextFile('data').map(lambda (name,text): text.split())
    tfVectors = tf.transform(rdd)    # 对整个RDD进行转换

# 在python中使用TF-IDF
    
    from pyspark.mllib.feature import Hashing,IDF
    
    # 将若干文本文件读取为TF向量
    rdd = sc.wholeTextFile('data').map(lambda (name,text):text.split())
    tf = HashingTF()
    tfVectors = tf.transform(rdd).cache()  # 缓存是因为它被使用了两次，一次是训练IDF模型，一次是用IDF乘TF向量的时候
    
    # 计算IDF,然后计算TF-IDF向量
    idf = IDF()
    idfModel = idf.fit(tfVectors)
    tfIdfVectors = idfModel.transform(tfVectors)

### 缩放 ###

将特征缩放到0均值，1方差的区间等。

from pyspark.mllib.feature import StandardScaler
    
    vectors = [Vectors.dense([-2.0, 5.0, 1.0]), Vectors.dense([2.0, 0.0, 1.0])]
    dataset = sc.parallelize(vectors)
    scaler = StandardScaler(withMean=True,withStd=True)
    model = scaler.fit(dataset)
    result = model.transform(dataset)

### 正则化 ###

默认情况下，使用L2欧几里得正则化，当然也可以使用其他范数正则化。

from pyspakr.mllib.feature import Normalizer
    normal = Normalizer()
    normal.fit(rdd)

### word2Vec ###

Word2Vec是一个基于神经网络的文本特征话方法，可以用来将数据传给许多下游算法。spark在`mllib.feature.Word2Vec`类中引入了该算法的一个实现。  
要训练word2vec，需要传给它一个用string类（每个单词一个）的iterable表示的语料库。类似地，你需要对语料库单词进行正规化处理（转为小写，取出标点，数字等）。注意，word2vec算法中模型的大小等于词库中单词数乘以向量的大小，向量大小默认为100.  
一般来说，比较合适的词库大小为10万个词。

## 统计 ##

`mllib.stat.Statistics`类中的方法提供了几种广泛使用的统计函数，这些函数可以直接使用在RDD上。

Statistics.colStats(rdd)
    # 计算由向量组成的RDD的统计性综述，包括每列的最大最小值，均值方差等
    
    Statistics.corr(rdd,mothod)
    # 计算由向量组成的RDD中的列间的相关矩阵，使用皮尔森或斯皮尔曼相关系数。mothod必须是pearson或者spearman中的一个。
    
    Statistics.corr(rdd1,rdd2,mothod)
    # 计算两个rdd的相关系数矩阵
    
    Statistics.chiSqTest(rdd)
    # 计算由LabeldPont对象组成的RDD中每个特征与标签的皮尔森独立性检验结果（卡方检验），返回的是一个ChiSqTestResult对象，其中有p值，测试统计及每个特征的自由度，标签和特征值必须是分类的（离散的）。
    
    # 其他统计函数
    mean()
    stdev()
    sum()
    sample()
    sampleByKey()

## 分类与回归 ##

分类和回归都会用到MLlib中的LabeldPoint类 。  
对于2分类，mllib的预期标签是0或1，在某些情况下，可能会使用-1和1，但这可能会导致错误的结果。对于多元分类，mllib的预期标签范围是0到C-1，C是类别的数目。

### 线性回归 ###

mllib.regression.LinearRegressionWithSGD
    # 随机梯度下降的普通线性回归
    
    LassoWithSGD
    # L1正则化的回归
    
    RidgeRegressionWithSGD
    # L2正则化的回归
    
    # 几个参数
    # numIterations 迭代次数
    # stepSize 梯度下降的步长，默认是1.0
    # intercept 是否给数据加上一个干扰项或者偏差特征，也就是一个值始终是1的特征，默认False
    # regParam lasso和ridge回归的正则化参数，默认是1.0

from pyspark.mllib.regression import LabeldPoint
    from pyspark.mllib.regression import LinearRegressionWithSGD
    
    points = # (创建labeldPoint组成的rdd)
    model = LinearRegressionWithSGD.train(points,iterations=200, intercept=True)
    
    # 打印模型的权重和偏置项
    print model.weights
    print model.intercept
    
    # 预测
    model.predict()

### 逻辑回归 ###

见前面垃圾邮件分类的例子。  
逻辑回归的拟合方法有SGD和LBFGS，一般LBFGS是最好的选择。  
一般情况下，分类的阈值是0.5，可以通过`setThreshold()`改变阈值，也可以通过`clearThreshold()`去掉阈值，这样`predict`返回的就是概率值。

# SGD方法
    from pyspark.mllib.classification import LogisticRegressionWithSGD
    # LBFGS方法
    from pyspark.mllib.classification import LogisticRegressionWithLBFGS

### SVM ###

同逻辑回归的参数差不多，同样使用阈值的方法进行预测。

from pyspark.mllib.classification import SVMWithSGD

### 朴素贝叶斯 ###

朴素贝叶斯支持一个参数，lambda（python中是lambda\_），用来进行平滑化。  
朴素贝叶斯训练好的模型，除了使用predict预测分类外，还可以查看模型的两个参数：各个特征与分类的可能性矩阵theta（对C个分类和D个特征的情况，矩阵大小为C\*D），以及表示先验概率的C为向量pi。

from pyspark.mllib.classification import NaiveBayes

### 决策树与随机森林 ###

from pyspark.mllib.tree import DecisionTree
    # 分类树的训练方法
    DecisionTree.trainClassifier()
    # 回归树的训练方法
    DecisionTree.trainRegressor()
    
    # 参数
    # data，由LabledPoint组成的rdd
    
    # numClasses，分类的数目，仅在分类树中使用
    
    # impurity，节点不存度，对于分类可以是gini或entropy，回归则必须是variance
    
    # maxDepth，树的最大深度
    
    # maxBins，在构建各节点时将数据分到多少个箱子，推荐值是32
    
    # categricalFeatureInfo，一个映射表，用来指定哪些特征是分类的，以及它们各有多少个分类。例如特征1是一个标签0或1的二元特征，特征2是一个[0,1,2]的三元特征，那么就应该传递{1:2, 2:3}。如果没有特征是分类的，那么就传递一个空的映射表。
    
    # 训练完成的model的toDebugString()可以输出这个树，而且是可序列化的。

随机森林

from pyspark.mllib.tree import RandomForest
    # 分类树的训练方法
    RandomForest.trainClassifier()
    # 回归树的训练方法
    DecisionTree.trainRegressor()
    
    # 参数
    # numTrees：构建森林的树的数量，提高numTrees可以降低对训练数据过拟合的可能性。
    
    # featrueSubsetStrategy：在每个节点上作决定时需要考虑的特征数量，可以是auto，all，sqrt，log2，以及 oneThird，越大的值所花费的代价也越大。
    
    # seed：随机数种子。
    
    # 随机森林算法返回一个WeightdEnsembleModel对象，其中包含几个决策树（在wakHyPotheses字段中，权重由wakHyPotheses决定），可以对rdd或vector调用predict()，它也有一个toDebugString()方法，可以打印其中所有的树。

### 聚类 ###

KMean和KMean++算法，后者可以提供更好的初始化策略

from pyspark.mllib.cluster import KMeans
    
    # 参数
    # K：聚类数目
    # initializationMode: 初始化聚类中心的方法，可以是'k-means++'（默认）或者random。
    # maxIteration：最大迭代次数
    # runs：算法并发运行的数目
    
    # model.clusterCenter: 访问聚类中心，是一个向量的数组

### 协同过滤 ###

# 使用交替最小二乘法ALS
    from pyspark.recommendation.ALS
    # ALS会为每个用户和产品都设一个特征向量，这样用户向量和产品向量的点积就接近与他们的得分。
    # 参数
    # rank：使用的特征向量的大小，更大的特征向量可以产生更好的模型，但是计算代价也大
    # iterations: 迭代次数
    # lambda：正则化参数
    # alpha: 用来在隐式ALS中计算置信度的常量，默认是1.0
    # numUserBlocks，numProductBlock：切分用户和产品数据的块的数目，用来控制并行度。传入-1则MLlib自动决定
    
    # 要是有ALS，需要有一个mllib.recommendation.Rating对象组成的rdd，其中每个包含一个用户ID,一个产品ID，和一个评分（要么是显示评分，要么是隐式反馈）。
    # 实现过程中的一个挑战是，每个ID都需要一个32位的整型值，如果ID是更多的字符串或者数字，那么推荐使用ID的哈希值，即使有两个用户或者产品映射到同一个ID上，总体结果依然不错。还有一种方法是，broadcast()一张从产品ID到整型值的表，来给每个产品独特的ID.
    
    # ALS返回一个 atrixFactorizationModel 对象来表示结果，可以调用predict()来对一个由（userid,productid）对组成的RDD进行预测评分。也可以使用model.recommendProducts(userid,productid)来为一个给定用户找到最值得推荐的前n个产品。
    
    # 需要注意的是， atrixFactorizationModel 对象很大，为每个用户和产品都存储了一个向量，这样我们就不能把它存到磁盘上，然后在另一个程序中读回来。不过，可以吧模型中生成的特征向量RDD，也就是model.userFeatures 和 model.productFeatures 分别保存到分布式文件系统上。
    
    # ALS有两个变种，显示评分（默认）和隐式反馈（调用ALS.trainImplicit()而不是ALS.train()打开）。用于显示评分时，每个用户对于产品的评分需要是一个得分（1-5分之类的），而预测出来的评分也是得分。对于隐式反馈，每个评分代表的是用户会和给定产品发生交互的置信度（比如随着用户访问一个页面次数增加，评分也会提高），预测出来的也是置信度。

## 降维 ##

### 主成分分析 ###

import org.apache.spark.mllib.linalg.Matrix
    improt org.apache.spark.millib.linalg.distributed.RowMatirx
    
    var points: RDD[Vector]= //...
    var mat: RowMatrix = new RowMatrix(points)
    var pc: Matrix = mat.computePrincipalComponents(2)
    
    //将点投射到低维空间
    var projected = mat.multiply(pc).rows
    // 在投射出的二维数据上训练K-means
    var model = KMeans.train(projected, 10)

### 奇异值分解 ###

// 计算RowMatrix矩阵中前20个奇异值及其对应的奇异值向量
    var svd: singularValueDecomposition[RowMatrix, Maxtrix] = mat.computeSVD(20, computeU=true)
    var U: RowMatrix= svd.u  // U是一个分布式RowMatrix
    var s: Vector = svd.s    // 奇异值用一个局部稠密向量表示
    var V: Maxtrix = svd.V   // V是一个局部稠密向量

## 模型评估 ##

from pyspark.mllib import evaluation