PySpark MLlib

PySpark MLlib

机器学习是一种将数据和统计工具结合起来预测输出的数据分析技术。各种企业行业都使用这种预测来做出有利的决策。

PySpark提供了一个被称为mllib的API来处理机器学习。PySpark的mllib支持各种机器学习算法，如分类、回归、聚类、协同过滤和降维，以及底层的优化原语。下面给出了各种机器学习概念：

• 分类

pyspark.mllib库支持几种分类方法，如二分类、多类分类和回归分析。对象可以属于不同的类别。分类的目标是基于信息区分数据。 Random Forest、Naive Bayes、Decision Tree是分类中最常用的算法。

• 聚类

聚类是一种无监督的机器学习问题。当您不知道如何对数据进行分类时，我们需要算法来查找模式并相应地对数据进行分类。常用的聚类算法有K-means聚类、高斯混合模型、层次聚类。

• fpm

fpm指频繁模式匹配，用于挖掘各种项、项集、子序列或其他子结构。它主要用于大规模数据集。

• linalg

mllib.linalg工具用于线性代数。

• 推荐

它用于定义用于做出推荐的相关数据。它能够预测未来的偏好并推荐前几个物品。例如，在线娱乐平台Netflix拥有大量电影，有时人们在选择喜欢的物品时面临困难。这就是推荐在该领域中扮演重要角色的地方。

• mllib回归

回归用于找到变量之间的关系和依赖关系。它找出数据的每个特征之间的相关性并预测未来的值。

mllib包支持许多其他算法、类和函数。在这里，我们将了解pyspak.mllib的基本概念。

MLlib特点

PySpark mllib对于迭代算法非常有用。以下是其特点：

• 提取： 从“行”数据中提取特征。
• 转换： 用于缩放、转换或修改特征。
• 选择： 从更大的特征集中选择有用的子集。
• 局部敏感哈希： 将特征转换与其他算法结合起来。

让我们来看一下PySpark MLlib的主要库。

MLlib线性回归

线性回归用于找到变量之间的关系和依赖关系。考虑以下代码：

frompyspark.sql import SparkSession
spark = SparkSession.builder.appName('Customer').getOrCreate()
frompyspark.ml.regression import LinearRegression
dataset = spark.read.csv(r'C:\Users\DEVANSH SHARMA\Ecommerce-Customers.csv')
dataset.show(10)

输出：

+--------------------+--------------------+----------------+------------------+------------------+------------------+--------------------+-------------------+
|                 _c0|                 _c1|             _c2|               _c3|               _c4|               _c5|                 _c6|                _c7|
+--------------------+--------------------+----------------+------------------+------------------+------------------+--------------------+-------------------+
|               Email|             Address|          Avatar|Avg Session Length|       Time on App|   Time on Website|Length of Membership|Yearly Amount Spent|
|mstephenson@ferna...|835 Frank TunnelW...|          Violet| 34.49726772511229| 12.65565114916675| 39.57766801952616|  4.0826206329529615|  587.9510539684005|
|   stash|4547 Archer Commo...|       DarkGreen| 31.92627202636016|11.109460728682564|37.268958868297744|    2.66403418213262|  392.2049334443264|
|    stash|24645 Valerie Uni...|          Bisque|33.000914755642675|11.330278057777512|37.110597442120856|   4.104543202376424| 487.54750486747207|
|riverarebecca@gma...|1414 David Throug...|     SaddleBrown| 34.30555662975554|13.717513665142507| 36.72128267790313|   3.120178782748092|  581.8523440352177|
|mstephens@davidso...|14023 Rodriguez P...|MediumAquaMarine| 33.33067252364639|12.795188551078114| 37.53665330059473|   4.446308318351434|  599.4060920457634|
|alvareznancy@luca...|645 Martha Park A...|     FloralWhite|33.871037879341976|12.026925339755056| 34.47687762925054|   5.493507201364199|   637.102447915074|
|katherine20@yahoo...|68388 Reyes Light...|   DarkSlateBlue| 32.02159550138701|11.366348309710526| 36.68377615286961|   4.685017246570912|  521.5721747578274|
|  stash|Unit 6538 Box 898...|            Aqua|32.739142938380326| 12.35195897300293| 37.37335885854755|  4.4342734348999375|  549.9041461052942|
|vchurch@walter-ma...|860 Lee KeyWest D...|          Salmon| 33.98777289568564|13.386235275676436|37.534497341555735|  3.2734335777477144|  570.2004089636196|
+--------------------+--------------------+----------------+------------------+------------------+------------------+--------------------+-------------------+
only showing top 10 rows

在下面的代码中，我们导入 VectorAssembler 库来创建一个新的列 Independent feature:

frompyspark.ml.linalg import Vectors
frompyspark.ml.feature import VectorAssembler
featureassembler = VectorAssembler(inputCols = ["Avg Session Length","Time on App","Time on Website"],outputCol = "Independent Features")
output = featureassembler.transform(dataset)
output.show()

输出：

+------------------+
Independent Feature
+------------------+
|34.49726772511229 |
|31.92627202636016 |
|33.000914755642675|
|34.30555662975554 |
|33.33067252364639 |
|33.871037879341976|
|32.02159550138701 |
|32.739142938380326|
|33.98777289568564 |
+------------------+

z = featureassembler.transform(dataset)
finlized_data = z.select("Indepenent feature", "Yearly Amount Spent",)
z.show()

输出：

+--------------------++-------------------+
|Independent Feature | Yearly Amount Spent|
+--------------------++-------------------+
|34.49726772511229   | 587.9510539684005  |
|31.92627202636016   | 392.2049334443264  |
|33.000914755642675  | 487.5475048674720  |
|34.30555662975554   | 581.8523440352177  |
|33.33067252364639   | 599.4060920457634  |
|33.871037879341976  | 637.102447915074   |
|32.02159550138701   | 521.5721747578274  |
|32.739142938380326  | 549.9041461052942  |
|33.98777289568564   | 570.2004089636196  |
+--------------------++-------------------+

PySpark提供了 LinearRegression() 函数来预测任何给定数据集的结果。语法如下：

regressor = LinearRegression(featureCol = 'column_name1', labelCol = 'column_name2 ')

MLlib K-Mean Cluster

K-Mean聚类算法是最流行和常用的算法之一。它用于将数据点聚类到预定义数量的簇中。下面的示例展示了MLlib K-Means Cluster库的使用：

from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
# Loads data.
dataset = spark.read.format("libsvm").load(r"C:\Users\DEVANSH SHARMA\Iris.csv")
# Trains a k-means model.
kmeans = KMeans().setK(2).setSeed(1)
model = kmeans.fit(dataset)
# Make predictions
predictions = model.transform(dataset)
# Evaluate clustering by computing Silhouette score
evaluator = ClusteringEvaluator()
silhouette = evaluator.evaluate(predictions)
print("Silhouette with squared euclidean distance = " + str(silhouette))
# Shows the result.
centers = model.clusterCenters()
print("Cluster Centers: ")
for center in centers:
    print(center)

PySpark MLlib的参数

以下是PySpark MLlib的几个重要参数：

• Ratings

这是Ratings的RDD，或是(userID，productID，rating)元组。

• Rank

表示计算特征矩阵的等级（特征数量）。

• Iterations

表示ALS的迭代次数（默认值为5）。

• Lambda

是正则化参数（默认值为0.01）。

• Blocks

用于并行计算一些块的数量。

协同过滤（mllib.recommendation）

协同过滤是一种通常用于推荐系统的技术。该技术重点在于填充用户-项目的缺失条目。目前，spark.ml支持基于模型的协同过滤。在协同过滤中，用户和产品由一小组隐藏因子描述，可以用于预测缺失条目。

正则化参数的缩放

正则化参数regParam被缩放以解决最小二乘问题。最小二乘问题在以下两种情况下发生：用户在更新用户因子时生成的评分数量，或者产品在更新产品因子时接收的评分数量。

冷启动策略

在进行常见的预测问题时，使用ALS模型（Alternative Least Square Model）进行预测。当测试数据集中出现用户或项目的情况，在训练模型时可能不存在。它可以发生在以下两种情况下：

• 在预测中，模型未经训练的用户和项目没有评分历史（这称为冷启动策略）。
• 在交叉验证期间，将数据在训练集和评估集之间拆分。在评估集中遇到不在训练集中的用户和项目是很常见的。

#importing the libraries
frompyspark.ml.evaluation import RegressionEvaluator
frompyspark.ml.recommendation import ALS
frompyspark.sql import Row
no_of_lines = spark.read.text(r"C:\Users\DEVANSH SHARMA\MovieLens.csv").rdd
no_of_parts = no_of_lines.map(lambda row: row.value.split("::"))
ratingsRDD = no_of_lines.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),
                                     rating=float(p[2]), timestamp=long(p[3])))
ratings = spark.createDataFrame(ratingsRDD)
(training, test) = ratings.randomSplit([0.8, 0.2])

# Develop the recommendation model using ALS on the training data
# Note we set cold start strategy to make sure that we don't get NaN evaluation metrics.
als = ALS(maxIter=5, regParam=0.01, userCol="userId", itemCol="movieId", ratingCol="rating",
    coldStartStrategy="drop")
model = als.fit(training)

# Calculate the model by computing the RMSE on the test data
predictions = model.transform(test)
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
predictionCol="prediction")
rmse = evaluator.evaluate(predictions)
print("Root-mean-square error = " + str(rmse))

# Evaluate top 10 movie recommendations for each user
userRecs = model.recommendForAllUsers(10)
# Evaluate top 10 user recommendations for each movie
movieRecs = model.recommendForAllItems(10)
# Evaluate top 10 movie recommendations for a specified set of users
users = ratings.select(als.getUserCol()).distinct().limit(3)
userSubsetRecs = model.recommendForUserSubset(users, 10)
# Evalute top 10 user recommendations for a specified set of movies
movies = ratings.select(als.getItemCol()).distinct().limit(3)
movieSubSetRecs = model.recommendForItemSubset(movies, 10)