PySpark MLlib

2025 年 8 月 28 日 | 阅读 6 分钟

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

PySpark 提供了与机器学习相关的 API，称为 mllib。PySpark 的 mllib 支持各种机器学习算法，如分类、回归、聚类、协同过滤和降维，以及底层的优化原语。下面将介绍各种机器学习概念

分类

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

聚类

聚类是一种无监督机器学习问题。当您不知道如何对数据进行分类时，会使用它；我们需要算法来寻找模式并据此对数据进行分类。流行的聚类算法是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|
|   [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|
|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|
|  [email protected]|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 库来创建一个新的独立特征列

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-均值聚类

K-均值聚类算法是最流行和最常用的算法之一。它用于将数据点聚类到预定义的簇数中。下面的示例展示了 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 的 RDD 或 (userID, productID, rating) 元组。

排名

它代表计算出的特征矩阵的秩（特征数）。

迭代

它代表 ALS 的迭代次数。（默认：5）

Lambda

它是正则化参数。（默认：0.01）

它用于并行化计算一定数量的块。

协同过滤 (mllib.recommendation)

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

正则化参数的缩放

正则化参数 regParam 被缩放以求解最小二乘问题。当用户对更新用户因子中的评分数量，或产品对更新产品因子中的评分数量时，会出现最小二乘问题。

冷启动策略

ALS 模型（交替最小二乘模型）用于在处理常见的预测问题时进行预测。在测试数据集中出现用户或项目而在训练模型时不存在时，会遇到这个问题。它可能出现在以下两种情况中

在预测中，模型没有为没有评分历史的用户和项目进行训练（称为冷启动策略）。
在交叉验证期间，数据在训练集和评估集之间进行划分。在评估集中遇到训练集中不存在的用户和项目是很普遍的。

让我们看下面的例子，我们从 MovieLens 数据集中加载评分数据。每一行包含一个用户、一部电影、评分和一个时间戳。

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

下一主题Python 网络爬虫

PySpark MLlib

MLlib 特性

MLlib 线性回归

MLlib K-均值聚类

PySpark MLlib 的参数

协同过滤 (mllib.recommendation)

正则化参数的缩放

冷启动策略

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Python教程

Python变量和数据类型

Python控制语句

Python数据结构

Python函数

Python模块

Python OOP

Python异常处理

Python文件处理

Python搜索和排序

Python高级主题

Python MySQL

Python MongoDB

Python SQLite

Python MCQ

Python Tkinter (GUI)

Python Web Blocker

Python内置函数

Python字符串函数

Python列表

Python字典

Plotly

相关教程

PySpark MLlib

MLlib 特性

MLlib 线性回归

MLlib K-均值聚类

PySpark MLlib 的参数

协同过滤 (mllib.recommendation)

正则化参数的缩放

冷启动策略

相关帖子

如何使用GEOPY计算两点之间的距离

Python魔术方法

使用Python进行网络抓取

Python高阶函数

Python IDE

Python SimpleImputer模块

Python正则表达式

Python数组

Python多处理

Python中的nsetools

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器