pyspark.mllib

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

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

• 分类

pyspark.mllib 库支持多种分类方法，如二元分类、多类分类和回归分析。该对象可能属于不同的类。分类的目的是根据信息区分数据。随机森林、朴素贝叶斯、决策树是分类中最有用的算法。

• 聚类

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

• fpm

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

• pyspark-mllib

mllib.linalg 实用程序用于线性代数。

• 推荐

用于定义推荐的相关数据。它能够预测未来的偏好，并推荐最热门的项目。比如在线娱乐平台网飞电影收藏量巨大，有时候人们在挑选喜欢的项目时会面临困难。这是推荐发挥重要作用的领域。

• mllib 回归

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

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

MLlib 功能

PySpark mllib 对迭代算法很有用。特点如下:

• **提取:**从“行”数据中提取特征。

• **变换:**用于缩放、转换或修改要素。

• **选择:**从更大的特征集中选择有用的子集。

• **位置敏感哈希:**它将特征变换的各个方面与其他算法相结合。

让我们来看看 PySpark 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|
|[email protected]|835 Frank TunnelW...|          Violet| 34.49726772511229| 12.65565114916675| 39.57766801952616|  4.0826206329529615|  587.9510539684005|
|   [email protected]|4547 Archer Commo...|       DarkGreen| 31.92627202636016|11.109460728682564|37.268958868297744|    2.66403418213262|  392.2049334443264|
|    [email protected]|24645 Valerie Uni...|          Bisque|33.000914755642675|11.330278057777512|37.110597442120856|   4.104543202376424| 487.54750486747207|
|riverar[email protected]|1414 David Throug...|     SaddleBrown| 34.30555662975554|13.717513665142507| 36.72128267790313|   3.120178782748092|  581.8523440352177|
|[email protected]|14023 Rodriguez P...|MediumAquaMarine| 33.33067252364639|12.795188551078114| 37.53665330059473|   4.446308318351434|  599.4060920457634|
|[email protected]|645 Martha Park A...|     FloralWhite|33.871037879341976|12.026925339755056| 34.47687762925054|   5.493507201364199|   637.102447915074|
|[email protected]|68388 Reyes Light...|   DarkSlateBlue| 32.02159550138701|11.366348309710526| 36.68377615286961|   4.685017246570912|  521.5721747578274|
|  [email protected]|Unit 6538 Box 898...|            Aqua|32.739142938380326| 12.35195897300293| 37.37335885854755|  4.4342734348999375|  549.9041461052942|
|[email protected]|860 Lee KeyWest D...|          Salmon| 33.98777289568564|13.386235275676436|37.534497341555735|  3.2734335777477144|  570.2004089636196|
+--------------------+--------------------+----------------+------------------+------------------+------------------+--------------------+-------------------+
only showing top 10 rows

在下面的代码中，我们正在导入 VectorAssembler 库来创建一个新的列独立特性:

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 提供了**线性回归()**函数来查找任何给定数据集的预测。语法如下:

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

K- Mean 聚类算法

K- Mean 聚类算法是最流行和最常用的算法之一。它用于将数据点聚类成预定义数量的聚类。下面的例子展示了 MLlib K-Means 集群库的使用:

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

它是 RDD 的评级或(用户标识，产品标识，评级)元组。

• Rank

它表示计算的特征矩阵的秩(特征的数量)。

• Iterations

它代表 ALS 的迭代次数。(默认值:5)

• Lambda

它是正则化参数。(默认值:0.01)

• Blocks

它用于并行计算一些块。

协同过滤

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

正则化参数的缩放

正则化参数正则化参数被缩放以解决最小二乘问题。最小二乘问题出现在用户在更新用户因素时生成的评分数，或产品在更新产品因素时收到的评分数。

冷启动策略

**ALS 模型(替代最小二乘模型)**用于预测，同时进行常见的预测问题。测试数据集中的用户或项目在训练模型期间可能不存在时遇到的问题。它可能发生在下面给出的两种情况下:

• 在预测中，模型不是针对没有评级历史的用户和项目进行训练的(称为冷启动策略)。

• 在交叉验证过程中，数据在训练集和评估集之间进行分割。遇到不在训练集中的评估集中的用户和项目是很普遍的。

Let’s consider the following example, where we load ratings data from the MovieLens dataset. Each row is containing a user, a movie, rating and a timestamp.

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

原文：https://www.javatpoint.com/pyspark-mllib

---