【头歌实训】Spark MLlib ( Python 版 )

文章目录

第1关：基本统计
- 编程要求
- 测试说明
- 答案代码
第2关：回归
- 编程要求
- 测试说明
- 参考资料
- 答案代码
第3关：分类
- 编程要求
- 测试说明
- 参考资料
- 答案代码
第4关：协同过滤
- 编程要求
- 测试说明
- 参考资料
- 答案代码
第5关：聚类
- 编程要求
- 测试说明
- 参考资料
- 答案代码
第6关：降维
- 编程要求
- 测试说明
- 参考资料
- 答案代码
第7关：特征提取与转化
- 编程要求
- 测试说明
- 答案代码
第8关：频繁模式挖掘
- 编程要求
- 测试说明
- 参考资料
- 答案代码
第9关：评估指标
- 编程要求
- 测试说明
- 答案代码

第1关：基本统计

编程要求

根据提示，在右侧编辑器补充代码，计算所给数据的 pearson 相关系数和 spearman 相关系数。

测试说明

平台会对你编写的代码进行测试：

预期输出：

DenseMatrix([[1.        , 0.05564149,        nan, 0.40047142],[0.05564149, 1.        ,        nan, 0.91359586],[       nan,        nan, 1.        ,        nan],[0.40047142, 0.91359586,        nan, 1.        ]])
DenseMatrix([[1.        , 0.10540926,        nan, 0.4       ],[0.10540926, 1.        ,        nan, 0.9486833 ],[       nan,        nan, 1.        ,        nan],[0.4       , 0.9486833 ,        nan, 1.        ]])

答案代码

from pyspark.ml.linalg import Vectors
from pyspark.ml.stat import Correlation
from pyspark.sql import SparkSessiondef trainingModel(spark):# 自定义数据集data = [(Vectors.sparse(4, [(0, 1.0), (3, -2.0)]),),(Vectors.dense([4.0, 5.0, 0.0, 3.0]),),(Vectors.dense([6.0, 7.0, 0.0, 8.0]),),(Vectors.sparse(4, [(0, 9.0), (3, 1.0)]),)]########## Begin ########### 将 data 转化为 DataFramedata = spark.createDataFrame(data, ['features'])# 计算 df 的 pearson 相关系数pearsonCorr = Correlation.corr(data, 'features', 'pearson').collect()[0][0]# 计算 df 的 spearman 相关系数spearmanCorr = Correlation.corr(data, 'features', 'spearman').collect()[0][0]# 返回 pearson 相关系数和 spearman 相关系数return pearsonCorr, spearmanCorr########## End ##########

第2关：回归

编程要求

根据提示，在右侧编辑器补充代码，实现线性回归的过程函数trainingModel(spark) ，函数返回训练好的模型。其中LinearRegression 只需设置以下三个参数：

maxIter=10
regParam=0.3
elasticNetParam=0.8

所需数据在 /data/workspace/myshixun/project/src/step2/linear.txt 中。

测试说明

平台会对你编写的代码进行测试，最终会输出该模型的 RMSE 指标:
$RMSE=\sqrt{\frac1m\sum_i(f(x_i)-y_i)^2}$
如果该指标在规定的范围内，则通过测试，测试代码将会输出 success，如果没有通过测试，将会输出 fail。预期输出： success

参考资料

Spark 官方文档

答案代码

from pyspark.ml.regression import LinearRegression
from pyspark.sql import SparkSessiondef trainingModel(spark):########## Begin ########### 读取数据data = spark.read.format("libsvm").load("/data/workspace/myshixun/project/src/step2/linear.txt")# 建立模型lr = LinearRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)# 训练模型model = lr.fit(data)# 返回模型return model########## end ##########

第3关：分类

编程要求

根据提示，在右侧编辑器补充代码，实现逻辑回归的过程函数trainingModel(spark) ，函数返回训练好的模型。其中LogisticRegression 只需设置以下三个参数：

maxIter=10
regParam=0.3
elasticNetParam=0.8

所需数据在 /data/workspace/myshixun/project/src/step3/logistic.txt 中。

测试说明

平台会对你编写的代码进行测试，最终会输出该模型的 roc 指标，如果该指标在规定的范围内，则通过测试，测试代码将会输出 success，如果没有通过测试，将会输出 fail。预期输出： success

参考资料

Spark 官方文档

答案代码

from pyspark.ml.classification import LogisticRegression
from pyspark.sql import SparkSessiondef trainingModel(spark):########## Begin ########### 读取数据data = spark.read.format("libsvm").load("/data/workspace/myshixun/project/src/step3/logistic.txt")# 建立模型lr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)# 训练模型model = lr.fit(data)# 返回模型,数据集return model########## End ##########

第4关：协同过滤

编程要求

根据提示，在右侧编辑器补充代码，实现协同过滤的过程函数trainingModel(spark) ，函数返回训练好的模型。所需数据在 /data/workspace/myshixun/project/src/step4/movie.txt 中，读取数据后请按如下要求命名列：

第一列：userID 数据类型：int
第二列：movieID 数据类型：int
第三列：rating 数据类型：float
第四列：timestamp 数据类型：int

其中ALS 除了要设置 userCol , itemCol 和 rating ，还需要设置以下三个参数：

maxIter=5
regParam=0.01
coldStartStrategy="drop"

测试说明

平台会对你编写的代码进行测试，最终会输出该模型的 RMSE 指标，如果该指标在规定的范围内，则通过测试，测试代码将会输出 success，如果没有通过测试，将会输出 fail。预期输出： success

参考资料

Spark 官方文档

答案代码

from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.recommendation import ALS
from pyspark.sql import SparkSession,Rowdef trainingModel(spark):########## Begin ########### 读取数据data = spark.read.text("/data/workspace/myshixun/project/src/step4/movie.txt")# 数据预处理，处理分隔符，为每一列添加索引data = data.rdd.map(lambda line: line.value.split('::'))\.map(lambda p: Row(userID=int(p[0]), movieID=int(p[1]), rating=float(p[2]), timestamp=int(p[3])))# 创建数据框ratings = spark.createDataFrame(data)# 划分训练集和测试集 8：2(train, test) = ratings.randomSplit([0.8, 0.2], seed=0)# 在训练集上使用 ALS 建立推荐系统als = ALS(maxIter=5, regParam=0.01, coldStartStrategy="drop", userCol="userID", itemCol="movieID", ratingCol="rating")# 训练模型model = als.fit(train)########## End ########### 计算测试集上的 RMSE 值predictions = model.transform(test)rmse = RegressionEvaluator(metricName="rmse", labelCol="rating",predictionCol="prediction").evaluate(predictions)# 返回 rmsereturn rmse

第5关：聚类

编程要求

根据提示，在右侧编辑器补充代码，实现聚类的过程函数trainingModel(spark) ，函数返回训练好的模型，对于模型你只需要设置以下两个参数：

k = 2
seed = 1

所需数据在 /data/workspace/myshixun/project/src/step5/k-means.txt 中

测试说明

平台会对你编写的代码进行测试，最终会输出该模型的 Silhouette score ，如果该指标在规定的范围内，则通过测试，测试代码将会输出 success，如果没有通过测试，将会输出 fail。预期输出： success

参考资料

Spark 官方文档

答案代码

from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
from pyspark.sql import SparkSessiondef trainingModel(spark):########## Begin ########### 读取数据data = spark.read.format("libsvm").load("/data/workspace/myshixun/project/src/step5/k-means.txt")# 建立 kmeans 模型kmeans = KMeans(k=2)kmeans.setSeed(1)# 训练模型model = kmeans.fit(data)########## End ##########predictions = model.transform(data)# 返回 模型、预测值return model, predictions

第6关：降维

编程要求

根据提示，在右侧编辑器补充代码，实现降维的过程函数trainingModel(spark) ，其中PCA 只需要设置以下三个参数：

k=3
inputCol="features", 
outputCol="pcaFeatures"

测试说明

平台会对你编写的代码进行测试，测试代码将会输训练好的矩阵。预期输出：

+-----------------------------------------------------------+
|pcaFeatures                                                |
+-----------------------------------------------------------+
|[1.6485728230883807,-4.013282700516296,-5.524543751369388] |
|[-4.645104331781534,-1.1167972663619026,-5.524543751369387]|
|[-6.428880535676489,-5.337951427775355,-5.524543751369389] |
+-----------------------------------------------------------+

参考资料

Spark 官方文档

答案代码

from pyspark.ml.feature import PCA
from pyspark.ml.linalg import Vectors
from pyspark.sql import SparkSessiondef trainingModel(spark):# 自定义数据集data = [(Vectors.sparse(5, [(1, 1.0), (3, 7.0)]),),(Vectors.dense([2.0, 0.0, 3.0, 4.0, 5.0]),),(Vectors.dense([4.0, 0.0, 0.0, 6.0, 7.0]),)]########## Begin ########### 创建数据框df = spark.createDataFrame(data, ["features"])# 建立模型pca = PCA(k=3, inputCol="features", outputCol="pcaFeatures")# 训练模型model = pca.fit(df)########## End ########### 返回计算结果result = model.transform(df).select('pcaFeatures')return result

第7关：特征提取与转化

编程要求

根据提示，在右侧编辑器补充代码，输出经过ml.feature.FeatureHasher 的特征，输出的特征命名为 features。注意，只需输出 feature 特征列。

测试说明

平台会对你编写的代码进行测试，测试代码将会输训练好的特征矩阵。预期输出：

+--------------------------------------------------------+
|features                                                |
+--------------------------------------------------------+
|(262144,[174475,247670,257907,262126],[2.2,1.0,1.0,1.0])|
|(262144,[70644,89673,173866,174475],[1.0,1.0,1.0,3.3])  |
|(262144,[22406,70644,174475,187923],[1.0,1.0,4.4,1.0])  |
|(262144,[70644,101499,174475,257907],[1.0,1.0,5.5,1.0]) |
+--------------------------------------------------------+

答案代码

from pyspark.ml.feature import FeatureHasher
from pyspark.sql import SparkSessiondef trainingModel(spark):# 自定义数据集data = spark.createDataFrame([(2.2, True, "1", "foo"),(3.3, False, "2", "bar"),(4.4, False, "3", "baz"),(5.5, False, "4", "foo")], ["real", "bool", "stringNum", "string"])########## Begin ########### 特征提取，使用 FeatureHasherhasher = FeatureHasher()hasher.setInputCols(["real", "bool", "stringNum", "string"])hasher.setOutputCol("features")hashed_df = hasher.transform(data)# 返回特征return hashed_df.select("features")########## End ##########

第8关：频繁模式挖掘

编程要求

根据提示，在右侧编辑器补充代码，实现频繁模式挖掘的过程函数trainingModel(spark) ，在设置 FPGrowth 只需要设置以下三个参数：

itemsCol="items", 
minSupport=0.5, 
minConfidence=0.6

测试说明

平台会对你编写的代码进行测试，最终会输出训练后的结果预期输出：

+---+------------+----------+
| id|       items|prediction|
+---+------------+----------+
|  0|   [1, 2, 5]|        []|
|  1|[1, 2, 3, 5]|        []|
|  2|      [1, 2]|       [5]|
+---+------------+----------+

参考资料

Spark 官方文档

答案代码

from pyspark.ml.fpm import FPGrowth
from pyspark.sql import SparkSessiondef trainingModel(spark):# 自定义数据集df = spark.createDataFrame([(0, [1, 2, 5]),(1, [1, 2, 3, 5]),(2, [1, 2])], ["id", "items"])########## Begin ########### 建立模型fp = FPGrowth(minSupport=0.5, minConfidence=0.6, itemsCol="items")# 训练模型fpm = fp.fit(df)# 返回模型,数据集return fpm, df########## End ##########

第9关：评估指标

编程要求

根据提示，在右侧编辑器补充代码，在逻辑回归实例中返回 areaUnderROC 指标和 acc 指标。

测试说明

平台会对你编写的代码进行测试，最终会返回逻辑回归实例中的 areaUnderROC 指标和 acc 指标，如果这两个值在设定范围内，将输出 success，否则输出 fail 。预期输出： success

答案代码

from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator
from pyspark.sql import SparkSessiondef trainingModel(spark):# 读取数据集data = spark.read.format("libsvm").load("/data/workspace/myshixun/project/src/step9/data.txt")# 建立模型lr = LogisticRegression(maxIter=10, regParam=0.3, elasticNetParam=0.8)# 训练模型model = lr.fit(data)########## Begin ########### 使用模型进行预测predictions = model.transform(data)# 创建 BinaryClassificationEvaluator 来计算 areaUnderROCevaluator_roc = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="label", metricName="areaUnderROC")areaUnderROC = evaluator_roc.evaluate(predictions)# 创建 MulticlassClassificationEvaluator 来计算 accuracyevaluator_acc = MulticlassClassificationEvaluator(predictionCol="prediction", labelCol="label", metricName="accuracy")accuracy = evaluator_acc.evaluate(predictions)# 返回 areaUnderROC 指标和 acc 指标return areaUnderROC, accuracy########## End ##########