【头歌实训】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 SparkSession

def 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 转化为 DataFrame
    data = 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 指标:
【头歌实训】Spark MLlib ( Python 版 )
如果该指标在规定的范围内，则通过测试，测试代码将会输出 success，如果没有通过测试，将会输出 fail。预期输出： success

参考资料

Spark 官方文档

答案代码

from pyspark.ml.regression import LinearRegression
from pyspark.sql import SparkSession

def 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 SparkSession

def 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,Row

def 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)

    # 返回 rmse
    return 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 SparkSession

def 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 SparkSession

def 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 SparkSession

def 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 ##########
    # 特征提取，使用 FeatureHasher
    hasher = 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 SparkSession

def 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 SparkSession

def 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 来计算 areaUnderROC
    evaluator_roc = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction", labelCol="label", metricName="areaUnderROC")
    areaUnderROC = evaluator_roc.evaluate(predictions)

    # 创建 MulticlassClassificationEvaluator 来计算 accuracy
    evaluator_acc = MulticlassClassificationEvaluator(predictionCol="prediction", labelCol="label", metricName="accuracy")
    accuracy = evaluator_acc.evaluate(predictions)
    
	# 返回 areaUnderROC 指标和 acc 指标
	return areaUnderROC, accuracy
    
    ########## End ##########

原文链接：https://blog.csdn.net/qq_61828116/article/details/135249861

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

文章目录

第1关：基本统计

编程要求

测试说明

答案代码

第2关：回归

编程要求

测试说明

参考资料

答案代码

第3关：分类

编程要求

测试说明

参考资料

答案代码

第4关：协同过滤

编程要求

测试说明

参考资料

答案代码

第5关：聚类

编程要求

测试说明

参考资料

答案代码

第6关：降维

编程要求

测试说明

参考资料

答案代码

第7关：特征提取与转化

编程要求

测试说明

答案代码

第8关：频繁模式挖掘

编程要求

测试说明

参考资料

答案代码

第9关：评估指标

编程要求

测试说明

答案代码

相关推荐