---

Win10下pyspark环境的配置请参考上篇博文：
【[2020.6] 数据挖掘：基于Spark框架的K-Means聚类1】Win10下配置Spark环境+Anaconda/Python安装pyspark，超详细

---

Iris（鸢尾花）数据集

Iris数据集简介

在博客一些用于聚类和分类问题的数据集里列举了一些常用的聚类和分类数据集。Iris 数据集是其中一种，既可用于聚类实验，也可用于分类实验。

Iris 数据集¹，也称鸢尾花数据集， 是一个经典数据分析数据集， 数据集包含 150 个数据样本，分为 3 类，每类 50 个数据，每个数据包含 4 个属性。可通过花萼长度，花萼宽度，花瓣长度，花瓣宽度 4 个属性预测鸢尾花卉属于（Setosa，Versicolour， Virginica）三个种类中的哪一类。 这里使用 Iris 数据集的属性特征进行聚类实验。

Iris数据集下载和处理

Iris plants 数据集可以从KEEL dataset或者UCI Machine Learning Repository下载，也可以直接从Sklearn.datasets机器学习包得到。

我选择从UCI Machine Learning Repository下载，点击 Data Folder，下载iris.data（实际是csv格式，逗号分隔的，可以用pandas包读取，代码如下）

url = "https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data"  
names = ['sepal-length', 'sepal-width', 'petal-length', 'petal-width', 'class']  
dataset = pandas .read_csv(url, names=names)
dataset.hist() #数据直方图histograms

这里因为我们要借助pyspark读取数据，所以对下载csv格式的iris.data先进行下格式上的预处理：

1. 将.data后缀修改为.text
2. 用excel打开iris.text，删除最后一行并保存，中间会让选择间隔方式什么的，直接默认，出来的text文件间隔应该是一个Tab

Iris 数据集的三维可视化结果如下图所示。

---

基于pypark的K-Means 聚类实验与参数分析

聚类实验的代码代码已经上传到github.

Spark组件MLlib

Spark生态系统已经成为伯克利数据分析软件栈BDAS（Berkeley Data Analytics Stack）的重要组成部分。BDAS的架构如下图所示，从中可以看出，Spark专注于数据的处理分析，而数据的存储还是要借助于Hadoop分布式文件系统HDFS、Amazon S3等来实现的。因此，Spark生态系统可以很好地实现与Hadoop生态系统的兼容，使得现有Hadoop应用程序可以非常容易地迁移到Spark系统中²。

Spark的生态系统主要包含了Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX 等组件，各个组件的具体功能如下：

• Spark Core：Spark Core包含Spark的基本功能，如内存计算、任务调度、部署模式、故障恢复、存储管理等。Spark建立在统一的抽象RDD之上，使其可以以基本一致的方式应对不同的大数据处理场景；通常所说的Apache Spark，就是指Spark Core；
• Spark SQL：Spark SQL允许开发人员直接处理RDD，同时也可查询Hive、HBase等外部数据源。Spark SQL的一个重要特点是其能够统一处理关系表和RDD，使得开发人员可以轻松地使用SQL命令进行查询，并进行更复杂的数据分析；
• Spark Streaming：Spark Streaming支持高吞吐量、可容错处理的实时流数据处理，其核心思路是将流式计算分解成一系列短小的批处理作业。Spark Streaming支持多种数据输入源，如Kafka、Flume和TCP套接字等；
• MLlib（机器学习）：MLlib提供了常用机器学习算法的实现，包括聚类、分类、回归、协同过滤等，降低了机器学习的门槛，开发人员只要具备一定的理论知识就能进行机器学习的工作；
• GraphX（图计算）：GraphX是Spark中用于图计算的API，可认为是Pregel在Spark上的重写及优化，Graphx性能良好，拥有丰富的功能和运算符，能在海量数据上自如地运行复杂的图算法。

本次实验主要用到的就是MLlib组件。

实验步骤与参数分析

1. 首先， 在 Machine Learning Repository 下载 Iris Dataset， 在 pyspark框架下读取数据集内容并将读取数据转为 DataFrame 格式，读取数据的结果如下图所示。
   

2. 通过设定不同的 k 值， 计算 K-means 聚类算法的聚类效果， 同时计算每次聚类结果下的聚类中心。 本实验设定 k 值为 2-9 之间的整数值。

3. 在 K-Means 算法的运算过程中， k 值的选择影响着聚类效果， k 值太小，容易将不是同一类的数据划分在一起， k 值太大，容易造成本是同一类的数据划分到不同的类中。可以通过肘部法则(Elbow Method)依托聚类性能评估选择适当的 k值，尽可能优化 K-Means 聚类效果。

肘部法则： K-Means 以最小化样本与质点平方误差作为目标函数，将每个簇的质点与簇内样本点的平方距离误差和称为畸变程度(distortions)，那么，对于一个簇，它的畸变程度越低，代表簇内成员越紧密，畸变程度越高，代表簇内结构越松散。畸变程度会随着类别的增加而降低，但对于有一定区分度的数据，在达到某个临界点时畸变程度会得到极大改善，之后缓慢下降，这个临界点就可以考虑为聚类性能较好的点。 这就是肘部法则的理论依据。

4. 可视化 K-Means 聚类实验中 k 值从 2 至 9 的畸变程度曲线，如下图 所示。
   
   当 k=6 时，可以认为此时的聚类性能较好。可视化 k=6 时的聚类结果，如下图所示。
   

---

实验代码

³

from pyspark import SparkContext
from pyspark.sql import SparkSession,Row
from pyspark.ml.clustering import KMeans
from pyspark.ml.linalg import Vectors
import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D


if __name__ == "__main__":
    sc = SparkContext(appName="KMeans_pyspark",master='local')  # SparkContext
    
    #----读取并处理数据----
    print("------------------读取数据-----------------")
    
    SparkSession(sc)   #利用SparkSession来使sc具有处理PipelinedRDD的能力
    
    def f(x):
        rel = {}
        rel['iris_features'] = Vectors.dense(float(x[0]),float(x[1]),float(x[2]),float(x[3]))
        return rel
    iris_DF = sc.textFile('./iris.txt').map(lambda line:line.split('\t')).map(lambda p:Row(**f(p))).toDF()  # transform RDD to DataFrame
    iris_DF.show()
        
    #----Kmeans聚类----
    print("------------------Kmeans聚类--------------------")
    print("------------设定不同的K值，进行分类,计算平方误差之和------------")
    
    errors = []
    results = []
    centers = []
    
    for k in range(2,10):
        kmeansmodel = KMeans().setK(k).setFeaturesCol('iris_features').setPredictionCol('prediction').fit(iris_DF)
   
        print("With K={}".format(k))
        
        #带有预测簇标签的数据集
        kmeans_results = kmeansmodel.transform(iris_DF).collect()
        results.append(kmeans_results)
        for item in kmeans_results:
            print(str(item[0])+' is predcted as cluster'+ str(item[1]))
        
        #获取到模型的所有聚类中心情况
        kmeans_centers = kmeansmodel.clusterCenters()
        centers.append(kmeans_centers)
        center_seq = 0
        for item in kmeans_centers:
            print("Cluster" +  str(center_seq) + "  Center" + str(item))
            center_seq = center_seq + 1
      
        #计算集合内误差平方和（Within Set Sum of Squared Error, WSSSE)
        WSSSE = kmeansmodel.computeCost(iris_DF)
        errors.append(WSSSE)
        print("Within Set Sum of Squared Error = " + str(WSSSE))
        
        print('--'*30 + '\n')
        
    #----WSSSE可视化----
    plt.figure()
    k_number = range(2,10)
    plt.plot(k_number,errors)
    plt.xlabel('Number of K')
    plt.ylabel('WSSSE')
    plt.title('K-WSSSE')
  
    
    #----聚类结果可视化----
    print("---------将数据转换为panda结构，并查看空间3d图心-----------")
     #通过K-WSSSE图，k=6时聚类效果较好
    k = 6

    cluster_vis = plt.figure(figsize=(10,10)).gca(projection='3d')
    
    for item in results[k-2]:
        if item[1] == 0:
            cluster_vis.scatter(item[0][0],item[0][1],item[0][2],c = 'b') # blue
        if item[1] == 1:
            cluster_vis.scatter(item[0][0],item[0][1],item[0][2],c = 'y') # yellow
        if item[1] == 2:
            cluster_vis.scatter(item[0][0],item[0][1],item[0][2],c = 'm') # magenta
        if item[1] == 3:
            cluster_vis.scatter(item[0][0],item[0][1],item[0][2],c = 'k') # black
        if item[1] == 4:
            cluster_vis.scatter(item[0][0],item[0][1],item[0][2],c = 'g') # green
        if item[1] == 5:
            cluster_vis.scatter(item[0][0],item[0][1],item[0][2],c = 'c') # cyan

   
    for item in centers[k-2]:
            cluster_vis.scatter(item[0],item[1],item[2],c = 'r',marker = 'p') # red,五角
             
    plt.show()

---

1. Fisher,R.A. The use of multiple measurements in taxonomic problems.Annual Eugenics, 7, Part II, 179-188 (1936) ↩︎

2. 子雨大数据之Spark入门教程:Spark简介 ↩︎

3. 子雨大数据之Spark入门教程:KMeans聚类算法 ↩︎