西瓜数据集（watermelon.txt）各个特征的含义如下： 数据集的每一行由3个数值组成，前2个数字用\t分隔，后2个数字用空格分隔。 对于数据集文件watermelon.txt，请编写MapReduce程序，同时采用密度和含糖率数据作为特征，设类别数为2，利用 K-Means 聚类方法通过多次迭代对数据进行聚类。不使用第三方库，选取合适的Spark RDD转换算子和行动算子实现Kmeans算法，完成实验内容； 5. 基于Spark MLlib，实现Kmeans聚类计算，利用idea写出完整代码以及所需的pom文件

由于题目中要求使用MapReduce程序实现K-Means算法，因此我们需要先将数据转换为key-value对的形式，以便于MapReduce程序的处理。我们可以将每个数据点看作是一个二维坐标系中的点，将其坐标作为key，将其所属的聚类中心作为value。具体实现如下：

1. 数据预处理

首先读取数据集文件watermelon.txt，将其中的每一行解析为一个二维坐标和所属聚类中心的编号。对于每个聚类中心，我们可以随机指定一个初始坐标，将其作为第一次迭代的聚类中心。

# 读取数据集文件
data = sc.textFile("watermelon.txt")

# 解析每个数据点
parsed_data = data.map(lambda line: tuple(map(float, line.split())))

# 随机初始化聚类中心
k = 2
centers = parsed_data.takeSample(False, k, 1)

2. K-Means算法迭代

接下来，我们可以使用MapReduce程序实现K-Means算法的迭代过程。在每次迭代中，我们需要对数据集中的每个点计算其与各个聚类中心的距离，并将其分配到距离最近的聚类中心所属的聚类中。然后，我们需要重新计算每个聚类中心的坐标，将其调整为该聚类中所有点的平均值。这样就完成了一次迭代，我们可以将新的聚类中心用于下一次迭代。

# 迭代次数
iterations = 10

for i in range(iterations):
    # 计算每个点与各个聚类中心的距离，将其分配到距离最近的聚类中心所属的聚类中
    cluster_assignment = parsed_data.map(lambda point: (closest_center(point, centers), point))
    
    # 计算新的聚类中心
    new_centers = cluster_assignment.groupByKey().mapValues(lambda points: average_points(points)).collect()
    
    # 更新聚类中心
    for center in new_centers:
        centers[center[0]] = center[1]

其中，closest_center函数用于计算每个点距离最近的聚类中心的编号，average_points函数用于计算一组点的平均值。

def closest_center(point, centers):
    """返回距离最近的聚类中心的编号"""
    closest_center = 0
    closest_distance = float('inf')
    for i in range(len(centers)):
        distance = euclidean_distance(point, centers[i])
        if distance < closest_distance:
            closest_distance = distance
            closest_center = i
    return closest_center

def average_points(points):
    """计算一组点的平均值"""
    num_points = len(points)
    if num_points == 0:
        return [0.0, 0.0]
    x = sum([point[0] for point in points]) / num_points
    y = sum([point[1] for point in points]) / num_points
    return [x, y]

def euclidean_distance(p1, p2):
    """计算两个点之间的欧几里得距离"""
    return ((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2) ** 0.5

3. 结果输出

最后，我们可以将聚类结果输出到文件中，以便于后续的分析和可视化。

# 将聚类结果输出到文件中
cluster_assignment.map(lambda pair: f"{pair[1][0]}\t{pair[1][1]}\t{pair[0]}").saveAsTextFile("output")

完整代码如下：

from pyspark import SparkContext

# 初始化SparkContext
sc = SparkContext()

# 解析数据集文件
data = sc.textFile("watermelon.txt")
parsed_data = data.map(lambda line: tuple(map(float, line.split())))

# 随机初始化聚类中心
k = 2
centers = parsed_data.takeSample(False, k, 1)

# 迭代次数
iterations = 10

for i in range(iterations):
    # 计算每个点与各个聚类中心的距离，将其分配到距离最近的聚类中心所属的聚类中
    cluster_assignment = parsed_data.map(lambda point: (closest_center(point, centers), point))
    
    # 计算新的聚类中心
    new_centers = cluster_assignment.groupByKey().mapValues(lambda points: average_points(points)).collect()
    
    # 更新聚类中心
    for center in new_centers:
        centers[center[0]] = center[1]

# 将聚类结果输出到文件中
cluster_assignment.map(lambda pair: f"{pair[1][0]}\t{pair[1][1]}\t{pair[0]}").saveAsTextFile("output")

def closest_center(point, centers):
    """返回距离最近的聚类中心的编号"""
    closest_center = 0
    closest_distance = float('inf')
    for i in range(len(centers)):
        distance = euclidean_distance(point, centers[i])
        if distance < closest_distance:
            closest_distance = distance
            closest_center = i
    return closest_center

def average_points(points):
    """计算一组点的平均值"""
    num_points = len(points)
    if num_points == 0:
        return [0.0, 0.0]
    x = sum([point[0] for point in points]) / num_points
    y = sum([point[1] for point in points]) / num_points
    return [x, y]

def euclidean_distance(p1, p2):
    """计算两个点之间的欧几里得距离"""
    return ((p1[0] - p2[0]) ** 2 + (p1[1] - p2[1]) ** 2) ** 0.5

这里仅提供了使用MapReduce程序实现K-Means算法的实现方式，如果需要使用Spark MLlib进行K-Means聚类计算，可以参考以下代码：

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

# 初始化SparkSession
spark = SparkSession.builder.appName("KMeans").getOrCreate()

# 读取数据集文件
data = spark.read.format("libsvm").load("watermelon.txt")

# 训练K-Means模型
kmeans = KMeans(k=2, seed=1)
model = kmeans.fit(data)

# 输出聚类结果
print("Cluster Centers:")
centers = model.clusterCenters()
for center in centers:
    print(center)

其中，libsvm格式的数据集文件是指每行数据的格式为"标签 特征1:值1 特征2:值2 ..."，在这里我们可以将标签设为0，将密度和含糖率作为两个特征。在输出聚类结果时，我们可以通过model.clusterCenters()获取聚类中心的坐标，从而进行后续分析和可视化。