第15章 大数据与MapReduce

假如你为一家网络购物商店工作，很多用户访问该网站，其中有些人会购买商品，有些人则随意浏览后就离开。
对于你来说，可能很想识别那些有购物意愿的用户。
那么问题就来了，数据集可能会非常大，在单机上训练要运行好几天。
接下来: 我们讲讲 MapRedece 如何来解决这样的问题

Hadoop 是 MapRedece 框架的一个免费开源实现。
MapReduce: 分布式的计算框架，可以将单个计算作业分配给多台计算机执行。

优点: 使程序以并行的方式执行，可在短时间内完成大量工作。
缺点: 算法必须经过重写，需要对系统工程有一定的理解。
适用数据类型: 数值型和标称型数据。

cat inputFile.txt | python mapper.py | sort | python reducer.py > outputFile.txt

# 测试 Mapper
# Linux
cat data/15.BigData_MapReduce/inputFile.txt | python src/python/15.BigData_MapReduce/mrMeanMapper.py
# Window
# python src/python/15.BigData_MapReduce/mrMeanMapper.py < data/15.BigData_MapReduce/inputFile.txt

# 测试 Reducer
# Linux
cat data/15.BigData_MapReduce/inputFile.txt | python src/python/15.BigData_MapReduce/mrMeanMapper.py | python src/python/15.BigData_MapReduce/mrMeanReducer.py
# Window
# python src/python/15.BigData_MapReduce/mrMeanMapper.py < data/15.BigData_MapReduce/inputFile.txt | python src/python/15.BigData_MapReduce/mrMeanReducer.py

python src/python/15.BigData_MapReduce/mrMean.py < data/15.BigData_MapReduce/inputFile.txt > data/15.BigData_MapReduce/myOut.txt

# 测试 mrjob的案例
# 先测试一下mapper方法
# python src/python/15.BigData_MapReduce/mrMean.py --mapper < data/15.BigData_MapReduce/inputFile.txt
# 运行整个程序，移除 --mapper 就行
python src/python/15.BigData_MapReduce/mrMean.py < data/15.BigData_MapReduce/inputFile.txt

将 回归系数w 初始化为0
对每次批处理
    随机选择 k 个样本点(向量)
    对每个向量
        如果该向量被错分: 
            更新权重向量 w
    累加对 w 的更新

收集数据: 数据按文本格式存放。
准备数据: 输入数据已经是可用的格式，所以不需任何准备工作。如果你需要解析一个大规模的数据集，建议使用 map 作业来完成，从而达到并行处理的目的。
分析数据: 无。
训练算法: 与普通的 SVM 一样，在分类器训练上仍需花费大量的时间。
测试算法: 在二维空间上可视化之后，观察超平面，判断算法是否有效。
使用算法: 本例不会展示一个完整的应用，但会展示如何在大数据集上训练SVM。该算法其中一个应用场景就是本文分类，通常在文本分类里可能有大量的文档和成千上万的特征。

0.365032        2.465645        -1
-2.494175       -0.292380       -1
-3.039364       -0.123108       -1
1.348150        0.255696        1
2.768494        1.234954        1
1.232328        -0.601198       1

def loadDataSet(fileName):
    dataMat = []
    labelMat = []
    fr = open(fileName)
    for line in fr.readlines():
        lineArr = line.strip().split('\t')
        # dataMat.append([float(lineArr[0]), float(lineArr[1]), float(lineArr[2])])
        dataMat.append([float(lineArr[0]), float(lineArr[1])])
        labelMat.append(float(lineArr[2]))
    return dataMat, labelMat

def batchPegasos(dataSet, labels, lam, T, k):
    """batchPegasos()

    Args:
        dataMat    特征集合
        labels     分类结果集合
        lam        固定值
        T          迭代次数
        k          待处理列表大小
    Returns:
        w          回归系数
    """
    m, n = shape(dataSet)
    w = zeros(n)  # 回归系数
    dataIndex = range(m)
    for t in range(1, T+1):
        wDelta = mat(zeros(n))  # 重置 wDelta

        # 它是学习率，代表了权重调整幅度的大小。（也可以理解为随机梯度的步长，使它不断减小，便于拟合）
        # 输入T和K分别设定了迭代次数和待处理列表的大小。在T次迭代过程中，每次需要重新计算eta
        eta = 1.0/(lam*t)
        random.shuffle(dataIndex)
        for j in range(k):      # 全部的训练集  内循环中执行批处理，将分类错误的值全部做累加后更新权重向量
            i = dataIndex[j]
            p = predict(w, dataSet[i, :])              # mapper 代码

            # 如果预测正确，并且预测结果的绝对值>=1，因为最大间隔为1, 认为没问题。
            # 否则算是预测错误, 通过预测错误的结果，来累计更新w.
            if labels[i]*p < 1:                        # mapper 代码
                wDelta += labels[i]*dataSet[i, :].A    # 累积变化
        # w通过不断的随机梯度的方式来优化
        w = (1.0 - 1/t)*w + (eta/k)*wDelta             # 在每个 T上应用更改
        # print '-----', w
    # print '++++++', w
    return w