【机器学习】在大数据上使用PySpark进行K-Means
共 5985字,需浏览 12分钟
·
2021-04-25 11:06
编译 | VK
来源 | Towards Data Science
如果你不熟悉K Means聚类,我建议你阅读下面的文章。本文主要研究数据并行和聚类,大数据上的K-Means聚类。
https://towardsdatascience.com/unsupervised-learning-techniques-using-python-k-means-and-silhouette-score-for-clustering-d6dd1f30b660
关于聚类
聚类是一种无监督的学习技术,简而言之,你处理的是数据,没有任何关于目标属性或因变量的信息。
聚类的一般思想是在数据中发现一些内在的结构,通常被称为相似对象的簇。该算法研究数据以识别这些簇,使得簇中的每个成员更接近簇中的另一个成员(较低的簇内距离),而远离不同簇中的另一个成员(较高的簇间距离)。
聚类适合哪里?
你们大多数人都熟悉现实生活中的这些例子:
客户细分-广泛用于目标营销 图像分割-识别景观 推荐引擎
背景
K-Means聚类,使用欧氏距离形式的相似性度量,通常被称为分裂聚类或分区聚类。
K均值的基本思想是从每个数据点都属于一个簇,然后根据用户输入K(或聚类数)将它们分成更小的簇。每个簇都有一个称为质心的中心。质心总数总是等于K。该算法迭代地寻找数据点并将它们分配给最近的簇。
一旦所有数据点被分配到各自的质心(这里代表每个簇),质心值将被重新计算,过程将重复,直到簇达到收敛标准。
质心只不过是每个簇的新平均值(例如,由客户A、B、C组成的簇,平均支出为100、200、300,篮子大小为10、15和20,质心分别为200和15)。收敛准则是衡量簇的稳定性的一个指标,即任意两次迭代之间的簇内距离在给定的阈值范围内不变。
Pypark有什么不同吗
在我们讨论为什么PySpark不是基于Sklearn的算法之前,让我们先讨论一下PySpark中的过程有什么不同。
在使用PySpark构建任何聚类算法时,都需要执行一些数据转换。让我们先理解数据,用于分析的数据可以在这里找到。
https://www.kaggle.com/arjunbhasin2013/ccdata
数据
该数据集由超过6个月的9K名活跃信用卡持卡人及其交易和账户属性组成。其想法是制定一个客户细分的营销策略。
使用Pypark
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName(‘Clustering using K-Means’).getOrCreate()
data_customer=spark.read.csv('CC General.csv', header=True, inferSchema=True)
data_customer.printSchema()
属性可以分为三大类。客户信息(主键为CUST_ID)、帐户信息(余额、余额频率、购买、信用额度、使用期限等)和交易(购买频率、付款、预付现金等)。
data_customer=data_customer.na.drop()
所考虑的所有属性都是数字或离散数字,因此我们需要使用向量汇编器(Vector Assembler)将它们转换为特征。向量汇编器是一种转换器,它将一组特征转换为单个向量列,通常称为特征数组,这里的特征是列。
customer id不会用于聚类。我们首先使用.columns提取所需的列,将其作为输入传递给Vector Assembler,然后使用transform将输入列转换为一个称为feature的向量列。
from pyspark.ml.feature import VectorAssembler
data_customer.columns
assemble=VectorAssembler(inputCols=[
'BALANCE',
'BALANCE_FREQUENCY',
'PURCHASES',
'ONEOFF_PURCHASES',
'INSTALLMENTS_PURCHASES',
'CASH_ADVANCE',
'PURCHASES_FREQUENCY',
'ONEOFF_PURCHASES_FREQUENCY',
'PURCHASES_INSTALLMENTS_FREQUENCY',
'CASH_ADVANCE_FREQUENCY',
'CASH_ADVANCE_TRX',
'PURCHASES_TRX',
'CREDIT_LIMIT',
'PAYMENTS',
'MINIMUM_PAYMENTS',
'PRC_FULL_PAYMENT',
'TENURE'], outputCol='features')
assembled_data=assemble.transform(data_customer)
assembled_data.show(2)
既然所有的列都被转换成一个单一的特征向量,我们就需要对数据进行标准化,使它们具有可比的规模。例如,BALANCE可以是10-1000,而BALANCE_FREQUENCY可以是0-1。
欧几里德距离总是在大尺度上受到更大的影响,因此对变量进行标准化是非常重要的。
from pyspark.ml.feature import StandardScaler
scale=StandardScaler(inputCol='features',outputCol='standardized')
data_scale=scale.fit(assembled_data)
data_scale_output=data_scale.transform(assembled_data)
data_scale_output.show(2)
既然我们的数据已经标准化了,我们就可以开发K均值算法了。
K-means是最常用的聚类算法之一,用于将数据分簇到预定义数量的聚类中。
spark.mllib包括k-means++方法的一个并行化变体,称为kmeans||。KMeans函数来自pyspark.ml.clustering,包括以下参数:
k是用户指定的簇数 maxIterations是聚类算法停止之前的最大迭代次数。请注意,如果簇内距离的变化不超过上面提到的epsilon值,迭代将停止,而不考虑最大迭代次数 initializationMode指定质心的随机初始化或通过k-means||初始化(类似于k-means++) epsilon决定k-均值收敛的距离阈值 initialModel是一簇可选的群集质心,用户可以将其作为输入提供。如果使用此参数,算法只运行一次,将点分配到最近的质心
train(k=4, maxIterations=20, minDivisibleClusterSize=1.0, seed=-1888008604)是默认值。
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator
silhouette_score=[]
evaluator = ClusteringEvaluator(predictionCol='prediction', featuresCol='standardized', \
metricName='silhouette', distanceMeasure='squaredEuclidean')
for i in range(2,10):
KMeans_algo=KMeans(featuresCol='standardized', k=i)
KMeans_fit=KMeans_algo.fit(data_scale_output)
output=KMeans_fit.transform(data_scale_output)
score=evaluator.evaluate(output)
silhouette_score.append(score)
print("Silhouette Score:",score)
可视化分数。注意,以前版本的K Means有computeScore,它计算聚类内距离的总和,但在spark3.0.0中被弃用。
轮廓分数使用ClusteringEvaluator,它测量一个簇中的每个点与相邻簇中的点的接近程度,从而帮助判断簇是否紧凑且间隔良好
# 可视化轮廓分数
import matplotlib.pyplot as plt
fig, ax = plt.subplots(1,1, figsize =(8,6))
ax.plot(range(2,10),silhouette_score)
ax.set_xlabel(‘k’)
ax.set_ylabel(‘cost’)
我更喜欢用K=7,在那里可以观察到轮廓分数的局部最大值。什么值的K是好的没有正确的答案。
我们可以使用描述性统计和其他图表来检查,这点在SkLearn和PCA上实现更方便。我们中的大多数人更喜欢研究肘部图,而不是轮廓分数,但PySpark有它的优点。
为什么是Pypark?
PySpark在执行K均值聚类时使用数据并行或结果并行的概念。
假设你需要为墨尔本节礼日活动开展有针对性的营销活动,并且你希望接触到具有不同购买属性的20万客户。想象一下在本地系统上运行K Means的多次迭代。对于K=5,需要计算的距离度量数为5 x 200K=1百万。100万个这样的度量需要计算30次才能满足收敛标准,即3000万个距离(欧几里德距离)。处理这样的场景需要大量的计算能力和时间。
数据并行性
数据并行所做的是,通过将数据集划分为更小的分区,从一开始就创建并行性。另一方面,结果并行是基于目标聚类的。例如:
D=记录数{X1,X2,…,Xn}
k=簇数
P=处理器数{P1,P2,…,Pm}
C=初始质心{C1,C2,…,Ck}
数据D被P个处理器分割。每个处理器处理一簇记录(由spark配置决定)。初始质心值C在每个处理器之间共享 现在每个处理器都有质心信息。处理器计算它们的记录到这些质心的距离,并通过将数据点分配到最近的质心来形成局部聚类 完成步骤2后,主进程将存储P个处理器上每个聚类的记录总数和计数,以供将来参考 一旦一次迭代完成,来自处理器的信息被交换,主进程计算更新的质心并再次在P个处理器之间共享它们,即,主进程更新质心,并与处理器重新共享信息 这个过程不断迭代直到收敛。一旦满足收敛条件,主进程就收集本地簇并将它们组合成一个全局聚类
想象一下,将20万条记录分成3个处理器,每个处理器有约70万条记录。这就是分布式处理的用武之地,以减少数据量,同时确保完整的结果。
结果并行性
例如:
D=记录数{X1,X2,…,Xn}
k=簇数
P=处理器数{P1,P2,…,Pm}
C=初始质心{C1,C2,…,Ck}
数据D被P个处理器分割,然后在每个处理器内排序。每个处理器处理一组记录(由spark配置决定) 初始质心值C被初始化,并在这些处理器中的每一个处理器之间进行分割/共享(即,与所有质心值在所有处理器之间共享的数据并行性不同,这里,我们将一个质心值传递给一个处理器) 现在每个处理器都有一个中心。计算这些点到这些质心的距离。对于处理器中的数据点:如果它们更接近处理器的质心,则将它们分配给该簇,否则如果它们更接近属于其他处理器的质心,则将数据点移动到新处理器 重复,直到收敛。
有用的链接
https://spark.apache.org/docs/latest/mllib-clustering.html https://spark.apache.org/docs/latest/api/python/pyspark.mllib.html#pyspark.mllib.clustering.KMeansModel https://spark.apache.org/docs/latest/api/python/pyspark.ml.html#pyspark.ml.evaluation.ClusteringEvaluator https://spark.apache.org/docs/latest/api/python/_modules/pyspark/ml/evaluation.html
感谢阅读。
往期精彩回顾
本站qq群851320808,加入微信群请扫码: