PySpark笔记(四)：MLlib包

MLlib是Spark中实现机器学习功能的模块，其主要针对RDD对象与DStream流对象。在Spark 2.0中，新引入的ML包是主要针对DataFrame对象的机器学习包。MLlib目前处于维护状态。

读入数据

首先读入数据，因为我们的数据以csv的形式保存，因此先以DataFrame的形式读入内存。

    sc = SparkSession.builder.master("local").appName("Demo")

    df = pd.read_csv("E:\Documents\Desktop\data.csv", encoding='utf-8')
    traffic = sc.createDataFrame(df).fillna(0)
    traffic.createOrReplaceTempView("traffic")

从数据中选取一个检测点作为我们要使用的点,可以看到该点共有2969条数据。数据读入时已经按照时间排序完成了，因此数据本身是保证了时序顺序的。

    b = traffic.filter(traffic.detectorid == 100625)

>>>b.count()
>>>2969
>>> b.show(5)
+----------+--------------+------+-----+---------+
|detectorid|     starttime|volume|speed|occupancy|
+----------+--------------+------+-----+---------+
|    100625|2015/12/1 0:00|    48|56.52|     1.29|
|    100625|2015/12/1 0:15|    50|53.54|     1.48|
|    100625|2015/12/1 0:30|    25|54.64|     0.62|
|    100625|2015/12/1 0:45|    34|54.94|     0.85|
|    100625|2015/12/1 1:00|    23|51.65|      0.6|
+----------+--------------+------+-----+---------+
only showing top 5 rows

描述性统计

首先选择出要使用的属性并将其转换为RDD。转换为RDD后，每个RDD保存原DataFrame每一行的数据。

>>> attr = ['volume', 'speed', 'occupancy']
>>> b.select(attr).show(5)

+------+-----+---------+
|volume|speed|occupancy|
+------+-----+---------+
|    48|56.52|     1.29|
|    50|53.54|     1.48|
|    25|54.64|     0.62|
|    34|54.94|     0.85|
|    23|51.65|      0.6|
+------+-----+---------+
only showing top 5 rows

>>> rb = b.select(attr).rdd.map(lambda row: [e for e in row])
>>> rb.take(5)
[[48, 56.52, 1.29], [50, 53.54, 1.48], [25, 54.64, 0.62], [34, 54.94, 0.85], [23, 51.65, 0.6]]

为筛选的列使用mlib的统计函数进行统计。(读入的时候需要对na值进行填充，否则包含na的列其统计信息也为na）

import pyspark.mllib.stat as st

stats = st.Statistics.colStats(rb)

for col, m, v in zip(attr, stats.mean(), stats.variance()):
    print('{0}: {1:.2f}, {2:.2f}'.format(col, m, v))

volume: 218.00, 14977.46
speed: 43.07, 209.62
occupancy: 14.19, 168.57

相关性

交通流数据之间具有较强的相关性，流量、速度、占有率之间可以按照特定的函数进行转化，我们使用mllib的相关性函数来查看三种属性之间的相关性。可以看出流量与占有率是正相关的，与速度是负相关的。

>>> coors = st.Statistics.corr(rb)
>>> coors

array([[ 1.        , -0.52093851,  0.63577129],
       [-0.52093851,  1.        , -0.92902088],
       [ 0.63577129, -0.92902088,  1.        ]])

回归预测

在这里我们使用随机森林进行回归预测，使用随机森林的好处是我们可以直接使用原始数据而不需要预处理。

首先我们读入数据，提取指定点的流量数据。

df = pd.read_csv("E:\Documents\Desktop\data.csv", encoding='utf-8')
volume = df[df.detectorid == 100625]['volume'].tolist()

然后我们根据时滞lag=4来构建二维的数据集合，即用过去四个点预测未来一个点。通过slide()函数我们可以获得一个大小为(n, 5)的矩阵。

def slide(data, lag):
    lag += 1
    res = []
    n = len(data)

    for i in range(lag, n):
        res.append(data[i - lag: i])

    return res

volume = slide(volume, lag)

然后我们将创建好的数据转化为RDD类型。

volume = sc.createDataFrame(volume)
volume = volume.rdd.map(lambda row: [e for e in row])

有了RDD类型之后，我们将其构建为Mllib中的模型能够使用的数据格式LabeledPoint。其参数如下，第一个值为预测值或者标签，后面的值为特征集合。

pyspark.mllib.regression.LabeledPoint(label, features)
Class that represents the features and labels of a data point.
| Parameters: |

• label – Label for this data point.
• features – Vector of features for this point (NumPy array, list, pyspark.mllib.linalg.SparseVector, or scipy.sparse column matrix).

labeled_v = volume.map(lambda row: LabeledPoint(row[-1], row[:-1]))

按照3：1划分训练集与测试集。

train, test = labeled_v.randomSplit([0.75, 0.25])

构造决策树模型并进行训练

model = DecisionTree.trainRegressor(train, {})

使用测试集进行预测，我们可以使用label或者features属性来访问LabeledPoint对象的属性。在这里预测出来的值要进行float转换，不换会出现TypeError: DoubleType can not accept object in type <type 'numpy.float64'>错误。

model = DecisionTree.trainRegressor(train, {})
y_pred = model.predict(test.map(lambda row: row.features))\
        .map(lambda row: float(row))

提取真实值，并将预测值与真实值配对，以方便送入评估器。

y_ture = test.map(lambda row: row.label)
# an RDD of (prediction, observation) pairs.
res = y_pred.zip(y_ture)
print(res.take(10))
eva(res)

Mllib提供了评估类用于评估模型效果，回归评估函数如下：

def eva(res):
    metrics = ev.RegressionMetrics(res)

    print("Explained Variance:{0:.2f}".format(metrics.explainedVariance))
    print("R2:{0:.2f}".format(metrics.r2))
    print("MAE:{0:.2f}".format(metrics.meanAbsoluteError))
    print("RMSE:{0:.2f}".format(metrics.rootMeanSquaredError))

预测结果：

将写好的文件提交到spark，运行结果如下：

res

完整代码：

import pandas as pd
from pyspark.sql import SparkSession
import pyspark.mllib.evaluation as ev
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.tree import DecisionTree


def slide(data, lag):
    lag += 1
    res = []
    n = len(data)

    for i in range(lag, n):
        res.append(data[i - lag: i])

    return res


def eva(res):
    metrics = ev.RegressionMetrics(res)

    print("Explained Variance:{0:.2f}".format(metrics.explainedVariance))
    print("R2:{0:.2f}".format(metrics.r2))
    print("MAE:{0:.2f}".format(metrics.meanAbsoluteError))
    print("RMSE:{0:.2f}".format(metrics.rootMeanSquaredError))


def main():
    sc = SparkSession.builder.master("local").appName("Demo").getOrCreate()

    lag = 4
    df = pd.read_csv("E:\Documents\Desktop\data.csv", encoding='utf-8')
    volume = df[df.detectorid == 100625]['volume'].tolist()
    volume = slide(volume, lag)
    volume = sc.createDataFrame(volume)
    volume = volume.rdd.map(lambda row: [e for e in row])

    labeled_v = volume.map(lambda row: LabeledPoint(row[-1], row[:-1]))
    train, test = labeled_v.randomSplit([0.75, 0.25])

    model = DecisionTree.trainRegressor(train, {})
    y_pred = model.predict(test.map(lambda row: row.features))\
        .map(lambda row: float(row))
    y_ture = test.map(lambda row: row.label)
    # an RDD of (prediction, observation) pairs.
    res = y_pred.zip(y_ture)
    print(res.take(10))
    eva(res)


if __name__ == '__main__':
    main()