1 0.697 0.460 2 0.774 0.376 3 0.634 0.264 4 0.608 0.318 5 0.556 0.215 6 0.403 0.237 7 0.481 0.149 8 0.437 0.211 9 0.666 0.091 10 0.243 0.267 11 0.245 0.057 12 0.343 0.099 13 0.639 0.161 14 0.657 0.198 15 0.360 0.370 16 0.593 0.042 17 0.719 0.103 18 0.359 0.188 19 0.339 0.241 20 0.282 0.257数据集的每一行由3个数值组成，前2个数字用\t分隔，后2个数字用空格分隔。 设类别数为2，利用 K-Means 聚类方法通过多次迭代对数据进行聚类。不使用第三方库，选取合适的Spark RDD转换算子和行动算子实现Kmeans算法，完成实验内容； 基于Spark MLlib，实现Kmeans聚类计算。代码实现

以下是使用Spark RDD实现的K-Means聚类算法的Python代码：

from pyspark import SparkContext, SparkConf
import math

def closest_centroid(datapoint, centroids):
    # 计算数据点距离最近的聚类中心
    closest = centroids[0]
    closest_distance = math.sqrt(
        (datapoint[0] - centroids[0][0]) ** 2 + (datapoint[1] - centroids[0][1]) ** 2
    )
    for centroid in centroids[1:]:
        distance = math.sqrt(
            (datapoint[0] - centroid[0]) ** 2 + (datapoint[1] - centroid[1]) ** 2
        )
        if distance < closest_distance:
            closest = centroid
            closest_distance = distance
    return closest

def kmeans_rdd(data, k, max_iterations):
    # 初始化聚类中心
    centroids = random.sample(data, k)
    for i in range(max_iterations):
        # 计算每个数据点距离最近的聚类中心
        closest = data.map(lambda p: (closest_centroid(p, centroids), p))
        # 将每个数据点分配到距离最近的聚类中心所在的簇
        assignments = closest.groupByKey()
        # 计算每个簇的平均值，得到新的聚类中心
        centroids = assignments.map(lambda x: (x[0], list(x[1]))).mapValues(
            lambda points: (
                sum(p[0] for p in points) / len(points),
                sum(p[1] for p in points) / len(points)
            )
        ).values().collect()
    # 返回每个数据点所属的簇
    return closest.map(lambda x: (x[1], x[0])).collect()

这个算法首先随机选择k个初始聚类中心，然后进行多次迭代。在每次迭代中，它首先计算每个数据点距离最近的聚类中心，然后将每个数据点分配到距离最近的聚类中心所在的簇。接着，它计算每个簇的平均值，得到新的聚类中心。重复进行多次迭代，直到聚类中心不再发生变化或达到最大迭代次数。最后，它返回每个数据点所属的簇。

以下是使用Spark MLlib实现的K-Means聚类算法的Python代码：

from pyspark.sql import SparkSession
from pyspark.ml.clustering import KMeans
from pyspark.ml.feature import VectorAssembler

def kmeans_mllib(data, k, max_iterations):
    # 将数据集转换成DataFrame
    spark = SparkSession.builder.appName("KMeans").getOrCreate()
    df = spark.createDataFrame(data, schema=["x", "y"])
    # 将DataFrame转换成向量
    assembler = VectorAssembler(inputCols=["x", "y"], outputCol="features")
    data = assembler.transform(df).select("features")
    # 创建KMeans模型并拟合数据集
    kmeans = KMeans(k=k, maxIter=max_iterations)
    model = kmeans.fit(data)
    # 将数据集中的每个向量分配到最近的中心点，并返回每个记录所属的簇
    predictions = model.transform(data)
    return predictions.rdd.map(lambda x: (x.features, x.prediction)).collect()

这个算法首先将数据集转换成DataFrame，然后使用`VectorAssembler`将数据集转换成一个向量。接着，它创建KMeans模型并使用`fit()`方法将数据集拟合到模型中。最后，它使用`transform()`方法将数据集中的每个向量分配到最近的中心点，并返回每个记录所属的簇。