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，运行结果如下：
完整代码：
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()
