西瓜数据集的k-means聚类算法python代码实现

由于题目中要求使用MapReduce编写K-Means算法，这里给出使用Hadoop Streaming实现的代码。 1. Mapper mapper读入每行数据，将密度和含糖率作为特征，输出键值对（簇序号，数据点信息）。 ```python #!/usr/bin/env python import sys # 读入聚类中心 centers = [] with open('centers.txt', 'r') as f: for line in f: center = line.strip().split('\t') centers.append((float(center[0]), float(center[1]))) # mapper for line in sys.stdin: data = line.strip().split('\t') x = float(data[0]) y = float(data[1]) min_dist = float('inf') cluster = -1 for i in range(len(centers)): center_x, center_y = centers[i] dist = (x - center_x) ** 2 + (y - center_y) ** 2 if dist < min_dist: min_dist = dist cluster = i print('{}\t{} {} {}'.format(cluster, x, y, data[2])) ``` 2. Reducer reducer读入每个簇的数据点信息，计算新的聚类中心，并输出键值对（新的簇序号，数据点信息）。 ```python #!/usr/bin/env python import sys # reducer cluster_dict = {} for line in sys.stdin: data = line.strip().split('\t') cluster = int(data[0]) x = float(data[1]) y = float(data[2]) info = data[3] if cluster not in cluster_dict: cluster_dict[cluster] = [(x, y)] else: cluster_dict[cluster].append((x, y)) for cluster in cluster_dict: center_x = sum([point[0] for point in cluster_dict[cluster]]) / len(cluster_dict[cluster]) center_y = sum([point[1] for point in cluster_dict[cluster]]) / len(cluster_dict[cluster]) print('{}\t{} {}\t{}'.format(cluster, center_x, center_y, len(cluster_dict[cluster]))) for point in cluster_dict[cluster]: print('{}\t{} {} {}'.format(cluster, point[0], point[1], info)) ``` 3. Driver driver程序用于多次迭代运行MapReduce程序，并将最终的聚类结果写入文件。 ```python #!/usr/bin/env python import os import shutil # 删除旧的输出目录 if os.path.exists('output'): shutil.rmtree('output') # 第一次迭代 os.system('hadoop jar /path/to/hadoop-streaming.jar \ -files mapper.py,reducer.py,centers.txt \ -input /path/to/watermelon.txt \ -output output/iter0 \ -mapper "python mapper.py" \ -reducer "python reducer.py"') # 迭代次数 iter_num = 10 # 迭代 for i in range(1, iter_num+1): # 更新聚类中心 os.system('hadoop fs -cat output/iter{}/part* > centers.txt'.format(i-1)) # 运行MapReduce程序 os.system('hadoop jar /path/to/hadoop-streaming.jar \ -D mapreduce.job.reduces=2 \ -files mapper.py,reducer.py,centers.txt \ -input /path/to/watermelon.txt \ -output output/iter{} \ -mapper "python mapper.py" \ -reducer "python reducer.py"'.format(i)) # 输出最终聚类结果 os.system('hadoop fs -cat output/iter{}/part* > result.txt'.format(iter_num)) ``` 4. 可视化 使用matplotlib库将聚类结果表示在二维平面上。 ```python import matplotlib.pyplot as plt x1, y1 = [], [] x2, y2 = [], [] with open('result.txt', 'r') as f: for line in f: data = line.strip().split('\t') x = float(data[1]) y = float(data[2]) cluster = int(data[0]) if cluster == 0: x1.append(x) y1.append(y) else: x2.append(x) y2.append(y) plt.scatter(x1, y1, s=50, c='r', marker='o', label='cluster1') plt.scatter(x2, y2, s=50, c='b', marker='o', label='cluster2') plt.legend() plt.show() ``` 完整代码：

由于题目中要求使用MapReduce程序实现K-Means算法，因此我们需要先将数据转换为key-value对的形式，以便于MapReduce程序的处理。我们可以将每个数据点看作是一个二维坐标系中的点，将其坐标作为key，将其所属的聚类中心作为value。具体实现如下： 1. 数据预处理 首先读取数据集文件watermelon.txt，将其中的每一行解析为一个二维坐标和所属聚类中心的编号。对于每个聚类中心，我们可以随机指定一个初始坐标，将其作为第一次迭代的聚类中心。 ```python # 读取数据集文件 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算法的迭代过程。在每次迭代中，我们需要对数据集中的每个点计算其与各个聚类中心的距离，并将其分配到距离最近的聚类中心所属的聚类中。然后，我们需要重新计算每个聚类中心的坐标，将其调整为该聚类中所有点的平均值。这样就完成了一次迭代，我们可以将新的聚类中心用于下一次迭代。 ```python # 迭代次数 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函数用于计算一组点的平均值。 ```python 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. 结果输出 最后，我们可以将聚类结果输出到文件中，以便于后续的分析和可视化。 ```python # 将聚类结果输出到文件中 cluster_assignment.map(lambda pair: f"{pair[1][0]}\t{pair[1][1]}\t{pair[0]}").saveAsTextFile("output") ``` 完整代码如下： ```python 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聚类计算，可以参考以下代码： ```python 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()获取聚类中心的坐标，从而进行后续分析和可视化。