<jupyter_start><jupyter_text>利用bike-sharing数据集构建回归模型 首先加载和查看数据集<jupyter_code>for line in open('data/Readme.txt').readlines():
    print line.strip()
rawdata = sc.textFile('data/hour.csv')
data = rawdata.map(lambda x:x.split(','))
for i in data.take(5):
    print i<jupyter_output>- instant: record index
- dteday : date
- season : season (1:springer, 2:summer, 3:fall, 4:winter)
- yr : year (0: 2011, 1:2012)
- mnth : month ( 1 to 12)
- hr : hour (0 to 23)
- holiday : weather day is holiday or not (extracted from http://dchr.dc.gov/page/holiday-schedule)
- weekday : day of the week
- workingday : if day is neither weekend nor holiday is 1, otherwise is 0.
+ weathersit :
- 1: Clear, Few clouds, Partly cloudy, Partly cloudy
- 2: Mist + Cloudy, Mist + Broken clouds, Mist + Few clouds, Mist
- 3: Light Snow, Light Rain + Thunderstorm + Scattered clouds, Light Rain + Scattered clouds
- 4: Heavy Rain + Ice Pallets + Thunderstorm + Mist, Snow + Fog
- temp : Normalized temperature in Celsius. The values are divided to 41 (max)
- atemp: Normalized feeling temperature in Celsius. The values are divided to 50 (max)
- hum: Normalized humidity. The values are divided to 100 (max)
- windspeed: Normalized wind speed. The values are divided to 67 (max)
- casual: count of casual us[...]<jupyter_text>可以看到第一行为标签，第一列instance为序号可以删除，第二列dteday为日期，信息已包含在之后的yr，mnth等变量中，可以删除。此次预测只预测总数量，因此去除casual和registered。剩下的变量分别为：季节，年，月，小时，节假日，星期几，工作日，天气，温度，体感温度，湿度，风速。其中后四个变量已标准化。<jupyter_code>data = data.filter(lambda x: x[0] != 'instant')
data.cache()
print data.count()<jupyter_output>17379<jupyter_text>我打算用线性回归和决策树两种方法来尝试拟合数据集，在线性回归中，对类型变量进行二元编码。（决策树中不需要）建立辅助函数并转换数据。<jupyter_code>def binary_encode(rdd, i):
    return (rdd.
           map(lambda x :x[i])
           .distinct()
           .zipWithIndex()
           .collectAsMap())
#对第2-9列应用该函数得到转换的字典
mappings = [binary_encode(data,i) for i in range(2,10)]
print mappings
#将第2-9列转换为二元编码
from pyspark.mllib.regression import LabeledPoint
import numpy as np
def extract_feature(line):
    lst = []
    for ind in range(2,10):
        dic = mappings[ind-2]
        for i,j in dic.iteritems():
            if line[ind] == i:
                lst.append(1.0)
            else:
                lst.append(0)
    for ind in range(10,14):
        lst.append(line[ind])
    return np.array(map(float,lst))
def extract_label(line):
    return float(line[-1])

binarydata = data.map(lambda x : LabeledPoint(extract_label(x), extract_feature(x)))
print binarydata.first()
print 'Raw data: '+ str(data.first()[2:])
print 'Raw Label: ' + str(data.first()[-1])
print 'Binary data: ' + str(binarydata.first().features)
print 'Binary label: '+ str(binarydata.first().label)
print 'Features number: ' + str(len(binarydata.first().features))<jupyter_output>[{u'1': 0, u'3': 1, u'2': 2, u'4': 3}, {u'1': 0, u'0': 1}, {u'11': 0, u'10': 1, u'12': 2, u'1': 3, u'3': 4, u'2': 5, u'5': 6, u'4': 7, u'7': 8, u'6': 9, u'9': 10, u'8': 11}, {u'20': 0, u'21': 1, u'22': 2, u'23': 3, u'1': 4, u'0': 5, u'3': 6, u'2': 7, u'5': 8, u'4': 9, u'7': 10, u'6': 11, u'9': 12, u'8': 13, u'11': 14, u'10': 15, u'13': 16, u'12': 17, u'15': 18, u'14': 19, u'17': 20, u'16': 21, u'19': 22, u'18': 23}, {u'1': 0, u'0': 1}, {u'1': 0, u'0': 1, u'3': 2, u'2': 3, u'5': 4, u'4': 5, u'6': 6}, {u'1': 0, u'0': 1}, {u'1': 0, u'3': 1, u'2': 2, u'4': 3}]
(16.0,[1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,0.0,0.0,0.0,1.0,0.0,1.0,1.0,0.0,0.0,0.0,0.24,0.2879,0.81,0.0])
Raw data: [u'1', u'0', u'1', u'0', u'0', u'6', u'0', u'1', u'0.24', u'0.2879', u'0.81', u'0', u'3', u'13', u'16']
Raw Label: 16
Binary data: [1.0,0.0,0.0,0.0,0.0,1.0,0.0,0.0,0.0,1[...]<jupyter_text>决策树中可以直接使用原始数据，因此只需要将u'1'格式转换为float，再用numpy封装一下<jupyter_code>def extract_float(line):
    return np.array(line[2:14])
treedata = data.map(lambda x: LabeledPoint(extract_label(x), extract_float(x)))
print 'Raw data: '+ str(data.first()[2:])
print 'Raw Label: ' + str(data.first()[-1])
print 'Desicion tree data: ' + str(treedata.first().features)
print 'Desicion tree Label: ' + str(treedata.first().label)<jupyter_output>Raw data: [u'1', u'0', u'1', u'0', u'0', u'6', u'0', u'1', u'0.24', u'0.2879', u'0.81', u'0', u'3', u'13', u'16']
Raw Label: 16
Desicion tree data: [1.0,0.0,1.0,0.0,0.0,6.0,0.0,1.0,0.24,0.2879,0.81,0.0]
Desicion tree Label: 16.0<jupyter_text>训练线性回归模型和决策树模型<jupyter_code>from pyspark.mllib.regression import LinearRegressionWithSGD

linear_model = LinearRegressionWithSGD.train(binarydata, iterations = 10, step = 0.1, intercept = False)
linear_pred_act = binarydata.map(lambda x: (linear_model.predict(x.features),x.label))
print 'linear predictions vs actual:'
print linear_pred_act.take(5)
from pyspark.mllib.tree import DecisionTree
tree_model = DecisionTree.trainRegressor(treedata,{})
preds = tree_model.predict(treedata.map(lambda x:x.features))
actual = treedata.map(lambda x: x.label)
tree_pred_act = preds.zip(actual)
print tree_pred_act.take(5)
print tree_model.depth()
print tree_model.numNodes()<jupyter_output>[(54.913223140495866, 16.0), (54.913223140495866, 40.0), (53.171052631578945, 32.0), (14.284023668639053, 13.0), (14.284023668639053, 1.0)]
5
63<jupyter_text>评估模型的性能，采用均方误差，均方根误差，平均绝对误差，R平方系数<jupyter_code>def squared_error(i,j):
    return (i-j)**2
def abs_error(i,j):
    return np.abs(i-j)
def square_log_error(i,j):
    return(np.log(i+1) - np.log(j+1))**2
mse = linear_pred_act.map(lambda (i,j): squared_error(i,j)).mean()
mae = linear_pred_act.map(lambda (i,j): abs_error(i,j)).mean()
rmsle = np.sqrt(linear_pred_act.map(lambda (i,j): square_log_error(i,j)).mean())
print 'mean squared error: '+ str(mse)
print 'mean absolute error: ' + str(mae)
print 'rot mean squred log error: ' + str(rmsle)
tree_mse = tree_pred_act.map(lambda (i,j): squared_error(i,j)).mean()
tree_mae = tree_pred_act.map(lambda (i,j): abs_error(i,j)).mean()
tree_rmsle = np.sqrt(tree_pred_act.map(lambda (i,j): square_log_error(i,j)).mean())
print 'desicion tree mean squared error: '+ str(tree_mse)
print 'desicion tree mean absolute error: ' + str(tree_mae)
print 'desicion tree rot mean squred log error: ' + str(tree_rmsle)
import matplotlib.pyplot as plt
targets = data.map(lambda x:float(x[-1])).collect()
plt.hist(targets,bins = 40,normed = True,color = 'lightblue')<jupyter_output><empty_output><jupyter_text>尝试取对数和平方根<jupyter_code>plt.hist(np.log(targets),bins = 40,normed = True,color = 'lightblue')
plt.hist(np.sqrt(targets),bins = 40,normed = True,color = 'lightblue')<jupyter_output><empty_output><jupyter_text>接下来尝试用对数拟合模型<jupyter_code>binarydata_log = binarydata.map(lambda x: LabeledPoint(np.log(x.label), x.features))
linear_model_log = LinearRegressionWithSGD.train(binarydata_log, iterations = 10, step = 0.1, intercept = False)
linear_pred_act_log = binarydata_log.map(lambda x: (np.exp(linear_model_log.predict(x.features)),np.exp(x.label)))
print 'linear predictions vs actual:'
print linear_pred_act_log.take(5)
mse = linear_pred_act_log.map(lambda (i,j): squared_error(i,j)).mean()
mae = linear_pred_act_log.map(lambda (i,j): abs_error(i,j)).mean()
rmsle = np.sqrt(linear_pred_act_log.map(lambda (i,j): square_log_error(i,j)).mean())
print 'mean squared error: '+ str(mse)
print 'mean absolute error: ' + str(mae)
print 'rot mean squred log error: ' + str(rmsle)
treedata_log = treedata.map(lambda x:LabeledPoint(np.log(x.label), x.features))
tree_model_log = DecisionTree.trainRegressor(treedata_log,{})
preds_log = tree_model_log.predict(treedata_log.map(lambda x:x.features)).map(lambda x: np.exp(x))
actual_log = treedata_log.map(lambda x: x.label).map(lambda x: np.exp(x))
tree_pred_act_log = preds_log.zip(actual_log)

tree_mse_log = tree_pred_act_log.map(lambda (i,j): squared_error(i,j)).mean()
tree_mae_log = tree_pred_act_log.map(lambda (i,j): abs_error(i,j)).mean()
tree_rmsle_log = np.sqrt(tree_pred_act_log.map(lambda (i,j): square_log_error(i,j)).mean())
print 'desicion tree mean squared error: '+ str(tree_mse_log)
print 'desicion tree mean absolute error: ' + str(tree_mae_log)
print 'desicion tree rot mean squred log error: ' + str(tree_rmsle_log)<jupyter_output>desicion tree mean squared error: 14781.5759883
desicion tree mean absolute error: 76.4131099112
desicion tree rot mean squred log error: 0.640599610072<jupyter_text>可以看到取对数对模型效果影响不大 创建训练集和测试集,测试不同参数对模型性能的影响<jupyter_code>data_with_index = binarydata.zipWithIndex().map(lambda (i,j): (j,i))
test = data_with_index.sample(False,0.2,42)
train = data_with_index.subtractByKey(test)
train_data = train.map(lambda (i,j):j)
test_data = test.map(lambda (i,j):j)
train_size = train_data.count()
test_size = test_data.count()
print "Training data size: %d" % train_size
print "Test data size: %d" % test_size
tree_data_with_index = treedata.zipWithIndex().map(lambda (i,j): (j,i))
tree_test = tree_data_with_index.sample(False,0.2,42)
tree_train = tree_data_with_index.subtractByKey(tree_test)
tree_train_data = train.map(lambda (i,j):j)
tree_test_data = test.map(lambda (i,j):j)
def evaluate(train,test,iterations ,step,regParam,regType,intercept):
    model = LinearRegressionWithSGD.train(train,iterations,step,regParam = regParam,regType = regType, intercept = intercept)
    tp = test.map(lambda x:(x.label,model.predict(x.features)))
    rmsle = np.sqrt(tp.map(lambda(i,j): square_log_error(i,j)).mean())
    return rmsle
params = [1,5,10,20,50,100,200,500,1000]
metrics = [evaluate(train_data,test_data,param,0.01,0.0,'l2',False) for param in params]
print params
print metrics
plt.plot(np.log(params),metrics)
params = [0.01,0.025,0.05,0.1,1]
metrics = [evaluate(train_data,test_data,10,param,0.0,'l2',False) for param in params]
print params
print metrics
plt.plot(np.log(params),metrics)
params = [0.0,0.01,0.1,1.0,5.0,10.0,20.0]
metrics = [evaluate(train_data,test_data,10,0.1,param,'l2',False) for param in params]
print params
print metrics
plt.plot(np.log(params),metrics)
params = [0.0,0.01,0.1,1.0,10.0,100.0,1000.0]
metrics = [evaluate(train_data,test_data,10,0.1,param,'l1',False) for param in params]
print params
print metrics
plt.plot(np.log(params),metrics)
params = [False,True]
metrics = [evaluate(train_data,test_data,10,0.1,1.0,'l1',param) for param in params]
print params
print metrics
plt.bar(params,metrics,color = 'lightblue')<jupyter_output>[False, True]
[1.466217949786879, 1.4931703614325664]<jupyter_text>可以发现线性回归模型的rmsle一直比较大，因此尝试优化决策树的参数<jupyter_code>def tree_evaluate(train,test,maxDepth,maxBins):
    model = DecisionTree.trainRegressor(train,{},impurity = 'variance', maxDepth = maxDepth, maxBins = maxBins)
    preds = model.predict(test.map(lambda x:x.features))
    actual = test.map(lambda x:x.label)
    tp = actual.zip(preds)
    rmsle = np.sqrt(tp.map(lambda (i,j): square_log_error(i,j)).mean())
    return rmsle
params = [1,2,3,4,5,10,20]
metrics = [tree_evaluate(tree_train_data,tree_test_data,param,32) for param in params]
print params
print metrics
plt.plot(params,metrics)
params = [2,4,8,16,32,64,100]
metrics = [tree_evaluate(tree_train_data,tree_test_data,20,param) for param in params]
print params
print metrics
plt.plot(params,metrics)<jupyter_output>[2, 4, 8, 16, 32, 64, 100]
[0.51801770741334463, 0.5392012822568224, 0.58125938230922403, 0.596482494988537, 0.63480636690000436, 0.62126914719040949, 0.6278350415424504]
{'fraction_non_alphanumeric': 0.1763931548924967, 'fraction_numerical': 0.08003510311540149, 'mean_word_length': 5.885800604229607, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '145483', 'license': 'MIT', 'n_tokens_mistral': 5426, 'n_tokens_neox': 5039, 'n_words': 908, 'path': 'usecases/bike-sharing/bike-sharing.ipynb', 'repo_name': 'superkley/spark-playground', 'size': '11395'}