如何为时间序列数据优化K-均值聚类速度？

在 New Relic，我们每分钟都会收集到 13.7 亿个数据点。我们为我们的客户收集、分析和展示的很大一部分数据都是时间序列数据。为了创建应用与其它实体（比如服务器和容器）之间的关系，以便打造 New Relic Radar 这样的新型智能产品，我们正在不断探索更快更有效的对时间序列数据分组的方法。鉴于我们所收集的数据的量是如此巨大，更快的聚类时间至关重要。

加速 k-均值聚类

k-均值聚类是一种流行的分组数据的方法。k-均值方法的基本原理涉及到确定每个数据点之间的距离并将它们分组成有意义的聚类。我们通常使用平面上的二维数据来演示这个过程。以超过二维的方式聚类当然是可行的，但可视化这种数据的过程会变得更为复杂。比如，下图给出了 k-均值聚类在两个任意维度上经过几次迭代的收敛情况：

不幸的是，这种方法并不能很好地用于时间序列数据，因为它们通常是随时间变化的一维数据。但是，我们仍然可以使用一些不同的函数来计算两个时间序列数据之间的距离因子（distance factor）。在这些案例中，我们可以使用均方误差（MSE）来探索不同的 k-均值实现。在测试这些实现的过程中，我们注意到很多实现的表现水平都有严重的问题，但我们仍然可以演示加速 k-均值聚类的可能方法，在某些案例中甚至能实现一个数量级的速度提升。

这里我们将使用 Python 的 NumPy 软件包。如果你决定上手跟着练习，你可以直接将这些代码复制和粘贴到 Jupyter Notebook 中。让我们从导入软件包开始吧，这是我们一直要用到的东西：

import timeimport numpy as npimport matplotlib.pyplot as plt %matplotlib inline

在接下来的测试中，我们首先生成 10000 个随机时间序列数据，每个数据的样本长度为 500。然后我们向随机长度的正弦波添加噪声。尽管这一类数据对 k-均值聚类方法而言并不理想，但它足以完成未优化的实现。

n = 10000ts_len = 500 phases = np.array(np.random.randint(0, 50, [n, 2]))pure = np.sin([np.linspace(-np.pi * x[0], -np.pi * x[1], ts_len) for x in phases])noise = np.array([np.random.normal(0, 1, ts_len) for x in range(n)]) signals = pure * noise # Normalize everything between 0 and 1signals += np.abs(np.min(signals))signals /= np.max(signals) plt.plot(signals[0])

第一个实现

让我们从最基本和最直接的实现开始吧。euclid_dist 可以为距离函数实现一个简单的 MSE 估计器，k_means 可以实现基本的 k-均值算法。我们从我们的初始数据集中选择了 num_clust 随机时间序列数据作为质心（代表每个聚类的中心）。在 num_iter 次迭代的过程中，我们会持续不断地移动质心，同时最小化这些质心与其它时间序列数据之间的距离。

def euclid_dist(t1, t2):        return np.sqrt(((t1-t2)**2).sum())def k_means(data, num_clust, num_iter):   centroids = signals[np.random.randint(0, signals.shape[0], num_clust)]    for n in range(num_iter):       assignments={}       for ind, i in enumerate(data):           min_dist = float('inf')           closest_clust = None           for c_ind, j in enumerate(centroids):               dist = euclid_dist(i, j)               if dist < min_dist:                  min_dist = dist                  closest_clust = c_ind            if closest_clust in assignments:               assignments[closest_clust].append(ind)           else:               assignments[closest_clust]=[]               assignments[closest_clust].append(ind)        for key in assignments:           clust_sum = 0           for k in assignments[key]:               clust_sum = clust_sum + data[k]           centroids[key] = [m / len(assignments[key]) for m in clust_sum]    return centroidst1 = time.time()centroids = k_means(signals, 100, 100)t2 = time.time()print("Took {} seconds".format(t2 - t1))Took 1138.8745470046997 seconds

聚类这些数据用去了接近 20 分钟。这不是很糟糕，但肯定算不上好。为了在下一个实现中达到更快的速度，我们决定去掉尽可能多的 for 循环。

向量化的实现

使用 NumPy 的一大优势是向量化运算。（如果你不太了解向量化运算，请参考这个链接：http://www.scipy-lectures.org/intro/numpy/operations.html）

k-均值算法要求每个质心和数据点都成对地进行比较。这意味着在我们之前的迭代中，我们要将 100 个质心和 10000 个时间序列数据分别进行比较，也就是每次迭代都要进行 100 万次比较。请记住每次比较都涉及到两个包含 500 个样本的集合。因为我们迭代了 100 次，那就是说我们总共比较了 1 亿次——对于单个 CPU 而言算是相当大的工作量了。尽管 Python 是一种还算高效的语言，但效率还赶不上用 C 语言写的指令。正是由于这个原因，NumPy 的大部分核心运算都是用 C 语言写的，并且还进行了向量化以最小化由循环带来的计算开销。

我们来探索一下我们可以如何向量化我们的代码，从而去掉尽可能多的循环。

首先，我们将代码分成不同的功能模块。这能让我们更好地理解每个部分所负责的工作。接下来，我们修改 calc_centroids 步骤以便仅在质心上迭代（而不是在每个时间序列数据上）。这样，我们将所有时间序列数据和一个质心传递给 euclid_dist。我们还可以预先分配 dist 矩阵，而不是将其当成一个词典进行处理并随时间扩展它。NumPy 的 argmin 可以一次性比较每个向量对。

在 move_centroids 中，我们使用向量运算去掉了另一个 for 循环，而且我们只在独特的质心集上迭代。如果我们丢失了一个质心，我们就通过从我们的时间序列数据集中进行随机选择来加入合适的数字（这在实际应用的实践中很罕见）。

最后，我们添加一个提前停止（early stopping）来检查 k_means——如果质心不再更新，就停止迭代。

来看看代码：

def euclid_dist(t1, t2):   return np.sqrt(((t1-t2)**2).sum(axis = 1)) def calc_centroids(data, centroids):   dist = np.zeros([data.shape[0], centroids.shape[0]])    for idx, centroid in enumerate(centroids):       dist[:, idx] = euclid_dist(centroid, data)    return np.array(dist) def closest_centroids(data, centroids):   dist = calc_centroids(data, centroids)   return np.argmin(dist, axis = 1) def move_centroids(data, closest, centroids):   k = centroids.shape[0]   new_centroids = np.array([data[closest == c].mean(axis = 0) for c in np.unique(closest)])    if k - new_centroids.shape[0] > 0:      print("adding {} centroid(s)".format(k - new_centroids.shape[0]))      additional_centroids = data[np.random.randint(0, data.shape[0], k - new_centroids.shape[0])]      new_centroids = np.append(new_centroids, additional_centroids, axis = 0)    return new_centroids def k_means(data, num_clust, num_iter):   centroids = signals[np.random.randint(0, signals.shape[0], num_clust)]   last_centroids = centroids    for n in range(num_iter):       closest = closest_centroids(data, centroids)       centroids = move_centroids(data, closest, centroids)       if not np.any(last_centroids != centroids):          print("early finish!")          break       last_centroids = centroids    return centroids

t1 = time.time()centroids = k_means(signals, 100, 100)t2 = time.time()print("Took {} seconds".format(t2 - t1))adding 1 centroid(s)early finish!took 206.72993397712708 seconds

耗时 3.5 分钟多一点。很不错！但我们还想完成得更快。

k-means++ 实现

我们的下一个实现使用了 k-means++ 算法。这个算法的目的是选择更优的初始质心。让我们看看这种优化方法有没有用……

def init_centroids(data, num_clust):   centroids = np.zeros([num_clust, data.shape[1]])   centroids[0,:] = data[np.random.randint(0, data.shape[0], 1)]    for i in range(1, num_clust):       D2 = np.min([np.linalg.norm(data - c, axis = 1)**2 for c in centroids[0:i, :]], axis = 0)       probs = D2/D2.sum()       cumprobs = probs.cumsum()       ind = np.where(cumprobs >= np.random.random())[0][0]       centroids[i, :] = np.expand_dims(data[ind], axis = 0)    return centroids def k_means(data, num_clust, num_iter):   centroids = init_centroids(data, num_clust)   last_centroids = centroids    for n in range(num_iter):       closest = closest_centroids(data, centroids)       centroids = move_centroids(data, closest, centroids)       if not np.any(last_centroids != centroids):           print("Early finish!")           break       last_centroids = centroids    return centroids

t1 = time.time()centroids = k_means(signals, 100, 100)t2 = time.time()print("Took {} seconds".format(t2 - t1))early finish!took 180.91435194015503 seconds

相比于我们之前的迭代，加入 k-means++ 算法能得到稍微好一点的性能。但是，当我们将其并行化之后，这种优化方法才真正开始带来显著回报。

 并行实现

到目前为止，我们所有的实现都是单线程的，所以我们决定探索 k-means++ 算法的并行化部分。因为我们在使用 Jupyter Notebook，所以我们选择使用用于并行计算的 ipyparallel 来管理并行性（ipyparallel 地址：https://github.com/ipython/ipyparallel）。使用 ipyparallel，我们不必担心整个服务器分叉，但我们需要解决一些特殊问题。比如说，我们必须指示我们的工作器节点加载 NumPy。

import ipyparallel as ippc = ipp.Client()v = c[:]v.use_cloudpickle() with v.sync_imports():   import numpy as np

关于加载工作器更多详情，请参阅 ipyparallel 上手指南：https://ipyparallel.readthedocs.io/en/latest/

在这一个实现中，我们的重点放在并行化的两个方面。首先，calc_centroids 有一个在每个质心上迭代并将其与我们的时间序列数据进行比较的循环。我们使用了 map_sync 来将这些迭代中的每一个发送到我们的工作器。

接下来，我们并行化 k-means++ 质心搜索中一个相似的循环。注意其中对 v.push 的调用：因为我们的 lambda 引用的数据，我们需要确保它在工作器节点上是可用的。我们通过调用 ipyparallel 的 push 方法来将该变量复制到工作器的全局范围中，从而实现了这一目标。

看看代码：

def calc_centroids(data, centroids):        return np.array(v.map_sync(lambda x: np.sqrt(((x - data)**2).sum(axis = 1)), centroids))def closest_centroids(points, centroids):   dist = calc_centroids(points, centroids)   return np.argmin(dist, axis=0) def init_centroids(data, num_clust):   v.push(dict(data=data))   centroids = np.zeros([num_clust, data.shape[1]])   centroids[0,:] = data[np.random.randint(0, data.shape[0], 1)]   for i in range(1, num_clust):       D2 = np.min(v.map_sync(lambda c: np.linalg.norm(data - c, axis = 1)**2, centroids[0:i,:]), axis = 0)       probs = D2/D2.sum()       cumprobs = probs.cumsum()       ind = np.where(cumprobs >= np.random.random())[0][0]       centroids[i, :] = np.expand_dims(data[ind], axis = 0)    return centroids

t1 = time.time()centroids = k_means(signals, 100, 100)t2 = time.time()print("Took {} seconds".format(t2 - t1))adding 2 centroid(s)early finish!took 143.49819207191467 seconds

结果只有两分钟多一点，这是我们目前实现的最快速度！

接下来：更快！

在这些测试中，我们都只使用了中央处理器（CPU）。CPU 能提供方便的并行化，但我们认为再多花点功夫，我们就可以使用图形处理器（GPU）来实现聚类，且速度将得到一个数量级的提升。我们也许可以使用 TensorFlow 来实现，这是一个用于数值计算和机器学习的开源软件。实际上，TensorFlow 已经包含了 k-均值实现，但我们基本上肯定还是需要对其进行调整才能将其用于时间序列聚类。不管怎样，我们都不会停下寻找更快更高效的聚类算法的步伐，以帮助管理我们的用户的数据。

生成海报

免责声明：本网站内容主要来自原创、合作伙伴供稿和第三方自媒体作者投稿，凡在本网站出现的信息，均仅供参考。本网站将尽力确保所提供信息的准确性及可靠性，但不保证有关资料的准确性及可靠性，读者在使用前请进一步核实，并对任何自主决定的行为负责。本网站对有关资料所引致的错误、不确或遗漏，概不负任何法律责任。任何单位或个人认为本网站中的网页或链接内容可能涉嫌侵犯其知识产权或存在不实内容时，应及时向本网站提出书面权利通知或不实情况说明，并提供身份证明、权属证明及详细侵权或不实情况证明。本网站在收到上述法律文件后，将会依法尽快联系相关文章源头核实，沟通删除相关内容或断开相关链接。