三维点云处理-聚类 3.2 KMeans 算法代码详细解释
·
一. KMeans
1. 概要
-
这份代码实现了一个 K-Means 聚类算法的类,使用 Python 编写。K-Means 是一种无监督学习算法,用于将数据点分成 K 个簇(clusters),每个簇由一个质心(centroid)表示。算法通过迭代优化质心位置来最小化簇内点到质心的距离平方和。
-
代码基于 NumPy 和 Pandas 库,支持随机初始化和 K-Means++ 初始化(一种更智能的初始化方法,能减少收敛到局部最优的概率)。

2. 其核心步骤包括:
-
人工指定聚类数量k,算法无法自动推断类别数
-
初始化阶段:随机选择k个中心点作为各类别的初始代表
-
迭代过程包含两个交替步骤:
- 数据点分配:计算每个点到k个中心的最近邻,确定类别归属
- 中心点更新:将每个类别的中心点重置为该类数据点的均值
3. 算法流程说明:
- 初始化:随机选择红蓝两中心点
- E步:按最近邻原则分配数据点类别
- M步:重新计算两类中心点为所属点均值
- 收敛条件包括:
- 中心点移动量小于阈值
- 类别分配不再变化
- 特殊情况下可能出现中心点震荡,需设置最大迭代次数终止

4. 应用技巧包括:
-
初始化优化:从数据点中选取初始中心而非完全随机
-
多轮运行:选择目标函数最小的结果作为最终输出
-
加速方法 :
- 使用kd-tree/octree加速最近邻搜索
- mini-batch变体:每次迭代使用数据子集降低计算量
二. 下面我逐段解释代码的含义,包括每部分的功能和逻辑。
1. 导入语句
import numpy as np
import pandas as pd
import random
import matplotlib.pyplot as plt
numpy as np: 用于数值计算,如数组操作、线性代数(e.g., 计算距离)。pandas as pd: 用于数据处理,将数据转换为 DataFrame 以便分组和聚合。random: 用于随机选择初始质心(在 K-Means++ 中)。matplotlib.pyplot as plt: 用于可视化结果(在主程序中绘制散点图)。
这些库是算法实现的基础,提供了高效的数据结构和计算工具。
2. 类定义和文档字符串
class KMeans(object):
"""
KMeans with both random and KMeans++ initialization
Parameters
----------
n_clusters: int
Number of clusters
tolerance: float
Initial splitting axis
max_iter: int
Maximum number of iterations
Attributes
----------
"""
- 定义了一个名为
KMeans的类,继承自object(Python 2 风格,但兼容 Python 3)。 - 文档字符串描述了类的参数:
n_clusters: 簇的数量(K 值)。tolerance: 容忍度,用于判断质心变化是否足够小以停止迭代(实际代码中是相对容忍度)。max_iter: 最大迭代次数,防止无限循环。
- 属性部分为空,但类中使用了私有属性(如
__K,__centroids)来存储状态。
3. 初始化方法 __init__
def __init__(self, n_clusters=2, tolerance=0.01, max_iter=300):
self.__K = n_clusters
self.__tolerance = tolerance
self.__max_iter = max_iter
self.__centroids = None
- 初始化类的实例变量:
self.__K: 存储簇的数量(私有属性,以双下划线开头)。self.__tolerance: 存储容忍度阈值。self.__max_iter: 存储最大迭代次数。self.__centroids: 初始化为None,将在fit方法中设置质心。
- 这是一个标准的构造函数,设置算法的参数。
4. fit 方法(训练算法)
def fit(self, data):
"""
Estimate the K centroids
Parameters
----------
data: numpy.ndarray
Training set as N-by-D numpy.ndarray
Returns
----------
None
"""
# TODO 01: implement KMeans fit
# get input size:
N, D = data.shape
# format as pandas dataframe:
__data = pd.DataFrame(
data = data,
index = np.arange(N),
columns = [f'x{i:03d}' for i in range(D)]
)
__data['cluster'] = 0
# get tolerance:
self.__tolerance = KMeans.__tolerance(data, tol)
# get initial centroids:
self.__centroids = self.__get_init_centroid_kmeanspp(data)
# iterate:
for i in range(self.__max_iter):
# expectation:
__data.cluster = __data.apply(
lambda x: KMeans.__assign(x[:-1].values, self.__centroids),
axis = 1
)
# maximization:
new_centroids = __data.groupby(['cluster']).mean().values
# evaluate squared diff:
diff = (new_centroids - self.__centroids).ravel()
squared_diff = np.dot(diff, diff)
# update centroids:
self.__centroids = new_centroids
# early stopping check:
if squared_diff <= self.__tolerance:
print(f'[KMeans - Fit]: early stopping with squared centroids diff {squared_diff:.2f} at iteration {i:03d}')
break
- 目的: 训练模型,估计 K 个质心的位置。
- 输入:
data是一个 N×D 的 NumPy 数组(N 个数据点,每个 D 维)。 - 步骤:
- 获取数据形状 (N, D)。
- 将数据转换为 Pandas DataFrame,便于操作(添加 ‘cluster’ 列初始化为 0)。
- 计算容忍度(基于数据方差的相对值)。
- 使用 K-Means++ 初始化质心(调用私有方法)。
- 迭代过程(EM 算法):
- 期望步 (Expectation): 为每个数据点分配最近的簇(使用
__assign方法)。 - 最大化步 (Maximization): 重新计算每个簇的质心(取簇内点的均值)。
- 计算质心变化的平方差(用于检查收敛)。
- 更新质心。
- 如果平方差小于容忍度,提前停止并打印信息。
- 期望步 (Expectation): 为每个数据点分配最近的簇(使用
- 这实现了标准的 K-Means 迭代,直到收敛或达到最大迭代。
5. predict 方法(预测簇标签)
def predict(self, data):
"""
Classify input data
Parameters
----------
data: numpy.ndarray
Testing set as N-by-D numpy.ndarray
Returns
----------
matches: numpy.ndarray
potential matches as N-by-2 numpy.ndarray
"""
# TODO 02: implement KMeans predict
N, _ = data.shape
result = np.asarray(
[KMeans.__assign(data[i], self.__centroids) for i in range(N)]
)
return result
- 目的: 为新数据点分配簇标签。
- 输入:
data是测试数据(N×D 数组)。 - 逻辑: 对每个数据点调用
__assign方法,找到最近的质心索引,返回一个长度为 N 的数组,表示每个点的簇 ID。 - 返回: 一个 NumPy 数组,包含每个数据点的簇标签(0 到 K-1)。
6. get_centroids 方法(获取质心)
def get_centroids(self):
"""
Get centroids
Parameters
----------
None
Returns
----------
centroids: numpy.ndarray
cluster centroids as numpy.ndarray
"""
return np.copy(self.__centroids)
- 目的: 返回当前质心的副本(避免外部修改)。
- 返回: K×D 的数组,每个行是一个质心。
7. 私有方法:初始化质心
def __get_init_centroid_random(self, data):
"""
Get initial centroids using random selection
Parameters
----------
data: numpy.ndarray
Training set as N-by-D numpy.ndarray
"""
N, _ = data.shape
idx_centroids = np.random.choice(np.arange(N), size=self.__K, replace=False)
centroids = data[idx_centroids]
return centroids
- 目的: 随机选择 K 个数据点作为初始质心(简单但可能导致局部最优)。
- 逻辑: 从 N 个点中无放回随机选择 K 个索引,返回对应的数据点。
def __get_init_centroid_kmeanspp(self, data):
"""
Get initial centroids using KMeans++ selection
Parameters
----------
data: numpy.ndarray
Training set as N-by-D numpy.ndarray
"""
N, _ = data.shape
# select the first centroid by random choice:
centroids = data[np.random.choice(np.arange(N), size=1, replace=False)]
# for the remaining centroids, select by prob based on minimum distance to existing centroids:
for _ in range(1, self.__K):
# find minimum distance to existing centroids for each poit
distances = np.asarray(
[
np.min(np.linalg.norm(d - centroids, axis=1))**2 for d in data
]
)
# generate cumulative probability:
probs = distances / np.sum(distances)
cum_probs = np.cumsum(probs)
# select new centroid:
centroids = np.vstack(
(centroids, data[np.searchsorted(cum_probs, random.random())])
)
return centroids
- 目的: 使用 K-Means++ 初始化,选择质心时考虑距离概率,更均匀分布。
- 逻辑:
- 第一个质心随机选择。
- 对后续质心:计算每个点到现有质心的最小距离平方,按距离加权选择(距离越远概率越高)。
- 使用累积概率和二分查找选择新质心。
8. 静态方法:辅助函数
@staticmethod
def __assign(data, centroids):
"""
Assign data point to centroids of minimum L2 distance
Parameters
----------
data: numpy.ndarray
Training set as N-by-D numpy.ndarray
centroids: numpy.ndarray
Centroids as N-by-D numpy.ndarray
"""
return np.argmin(np.linalg.norm(centroids - data, axis=1))
- 目的: 为单个数据点分配最近的簇。
- 逻辑: 计算点到所有质心的 L2 距离,返回最小距离的索引。
@staticmethod
def __tolerance(data, tol):
"""
Return a tolerance which is independent of the dataset
"""
variances = np.var(data, axis=0)
return np.mean(variances) * tol
- 目的: 计算相对容忍度(基于数据方差的平均值乘以用户指定的 tol)。
- 逻辑: 容忍度与数据集规模无关,避免不同尺度数据的问题。
9. 主程序(测试代码)
if __name__ == '__main__':
# create test set:
K = 2
X = np.array(
[
[1, 2],
[1.5, 1.8],
[5, 8],
[8, 8],
[1, 0.6],
[9, 11]
]
)
# fit:
k_means = KMeans(n_clusters=K)
k_means.fit(X)
# predict:
category = k_means.predict(X)
# visualize:
color = ['red','blue','green','cyan','magenta']
labels = [f'Cluster{k:02d}' for k in range(K)]
for k in range(K):
plt.scatter(X[category == k][:,0], X[category == k][:,1], c=color[k], label=labels[k])
centroids = k_means.get_centroids()
plt.scatter(centroids[:,0], centroids[:,1] ,s=300, c='grey', marker='P', label='Centroids')
plt.xlabel('X')
plt.ylabel('Y')
plt.legend()
plt.title('KMeans Testcase')
# plt.show()
plt.savefig("kmeans_result.png", dpi=200, bbox_inches="tight")
print("Saved figure to kmeans_result.png")
plt.close()
- 目的: 测试算法,使用简单 2D 数据集。
- 步骤:
- 创建测试数据(6 个点,2 个簇)。
- 训练模型,预测标签。
- 可视化:绘制散点图(按簇着色),标记质心,保存为 PNG 文件。
- 这是一个完整的示例,验证算法是否工作。
这个实现是标准的 K-Means,但有一些优化(如 K-Means++)。如果有特定部分需要更深入解释,请告诉我!
三. k-medoids
1.k-means的局限性
- 预设k值问题:聚类数量k需预先设定,实际应用中常依赖实验猜测,缺乏自动确定机制。
- 噪声敏感性:对异常点敏感,需通过k-medoids等改进算法缓解噪声影响。
- 硬分类缺陷:仅支持非概率性分类,边界点无法输出隶属概率(如50%属A类,50%属B类),此缺陷由高斯混合模型(GMM)弥补。

2.k-medoids
k-medoids改进动机:
| 对比维度 | k-means | k-medoids |
|---|---|---|
| 中心点性质 | 可不在数据集中 | 必须为实际数据点 |
| 抗噪能力 | 易受离群点影响 | 通过最小化距离和抵抗噪声 |
| 数据类型 | 需可计算均值 | 支持任意可定义距离的数据 |
1) k-medoids的estep
E步与k-means一致:通过预定义的距离函数v计算数据点到各中心点的距离,按最近邻原则分配类别。
2) k-medoids的mstep
M步采用枚举法:
- 对每个类别遍历其所有数据点
- 计算各点作为中心点时与类内其他点的距离和
- 选择使距离和最小的点作为新中心点
- 时间复杂度为O(nk²),其中nk为类内点数。
3) 图片压缩的应用

更多推荐


所有评论(0)