一、 实验内容 西瓜数据集（watermelon.txt）各个特征的含义如下： 数据集的每一行由3个数值组成，前2个数字用\t分隔，后2个数字用空格分隔。 对于数据集文件watermelon.txt，请编写MapReduce程序，同时采用密度和含糖率数据作为特征，设类别数为2，利用 K-Means 聚类方法通过多次迭代对数据进行聚类。输出聚类结果，包括数据点信息与对应簇序号，并观察输出结果是否正确； 9. 使用Python将聚类结果表示在二维平面上。写出完整代码

由于题目中要求使用MapReduce编写K-Means算法，这里给出使用Hadoop Streaming实现的代码。

1. Mapper

mapper读入每行数据，将密度和含糖率作为特征，输出键值对（簇序号，数据点信息）。

#!/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读入每个簇的数据点信息，计算新的聚类中心，并输出键值对（新的簇序号，数据点信息）。

#!/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程序，并将最终的聚类结果写入文件。

#!/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库将聚类结果表示在二维平面上。

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()

完整代码：