实际业务场景中，每天文本内容的数据量都在亿级以上，为了更高效地处理如此海量的数据，文本聚类技术的运用是必不可少的。
所谓文本聚类，指的是将文本按照类别进行聚合，然后以类别为单位对文本进行处理或使用。文本聚类之所以能够对文本按照类别进行聚类，主要是基于一个聚类假设：同类的文本相似度较大，而不同类的文本相似度较小。从上述文本聚类的定义看，文本聚类似乎与文本分类做的事情差不多，它们有什么不同吗？
与文本分类相比，文本聚类是一个无监督的算法，它既没有模型的训练过程，也不需要预先对文本进行人工标注类别。此外，文本分类的类别体系一般是事先定义好的，类别的总数量是固定的，并且每个类别都有清晰明确的实际意义。文本聚类的类别则是不固定的，完全取决于数据的实际分布情况，最后聚合出来的各个类别也没明确的定义，只是代表类别内的文本之间比较相似，与其他类别的文本比较不同。
易盾在内容安全领域深耕多年，对于文本类不良内容的识别与过滤已经发展出一套融合多类核心技术的成熟方案。易盾的文本内容安全方案里面综合使用了黑名单、关键词、规则、分类模型、特征库匹配、用户画像、高频监控、实时聚类等技术，这些技术各有其优势和局限性，只有灵活运用协同作战，才能发挥最佳的实际效果。
比如对于一些确定不是正常用户的账号，黑名单可以对其进行有力地封禁，彻底杜绝该账号的发垃圾行为。但如果是正常用户偶尔发表的一些不当内容，则不可以使用黑名单来处理。关键词比较适合用来拦截带有明显特征词语的内容。以一些色情词汇为例，这类内容使用关键词技术干脆利落，并且不会有漏掉的情况。而如果是一些比较内涵的内容，使用关键词就会存在大量的误判。其他的技术也是同样的情况，这里就不一一赘述了。
道高一尺魔高一丈，黑灰产与产品方的对抗在内容安全领域是一种常态，任何在线检测系统都无法保证不会漏过任何有害内容。我们需要建立一种能及时发现漏过的有害内容的机制，以此来保证在线检测效果的稳定性。
一般来说，对于在线检测未识别的内容，主要通过可疑审核、走查、聚类审核三种主要方法来召回漏过的有害内容，并同时维护对应的特征库，从而持续完善在线检测系统的识别能力。其中，可疑审核主要是通过一些高召回的手段，标记出未识别内容中哪些是比较可疑的，然后进行人工审核确认。质检人员会凭经验通过一些线索词，从未识别内容中搜索出一些内容进行走查确认，从中发现漏过的有害内容。作为文本的主角，实时聚类技术则对线上未识别的文本内容进行实时的聚合，把其中大量相似的内容同步到后台的审核系统，供审核人员人工审核，判断该内容是否有问题。
引入实时聚类技术用于未识别有害内容召回，主要是基于以下几方面的考量：
1）大部分有害内容为了增加曝光度，往往存在多次发布的行为，并且为了规避反垃圾系统的过滤，会对内容进行小幅度的修改和变化，聚类技术能够在海量的未识别内容当中把这批内容聚合到一起。
2）有害内容毕竟只占线上数据的一小部分，未识别内容里面绝大多数是正常内容，由于语言和表达的多样性，正常内容之间相似的情况较少，聚类能够排除掉大量的正常内容，减少人工审核量。
3）发布有害内容的黑产，一直在对内容进行修改，尝试绕过反垃圾小系统的识别，一旦成功了，就会使用程序的方式在短时间内容把大量的内容刷出去，造成是否恶劣的影响，聚类技术能够及时发现这种异常，从而控制有害漏过刷量的风险。
4）有些有害内容形式上比较隐蔽，单个内容上面看不出什么问题，需要全局考量，才能发现其中的规律。比如下面的内容，好像就是一个女生发的自拍照，似乎看不出什么问题。
但如果放到下面这批数据中，则可以看出这批数据呈现出一定规律，明显是有问题的。
一、文本聚类
文本聚类包含两个关键技术：
相似度量和聚合方法。
相似度量主要解决如下问题：给定两个文本，怎么判断它们是相似还是不相似的。
相似度量在数学上满足自反性和对称性，不满足传递性。满足自反性，即文本A永远和它自己相似。对称性指的是如果文本A与文本B相似，则可以得到文本B与文本A相似。假设文本A与文本B相似，文本B与文本C相似，我们无法确定文本A与文本C相似，这就是不满足传递性。聚合方法则解决的是下面的问题：给定一批文本，如何把相似的聚到同一个类别，不相似的聚到不同类别？
这个问题看起来似乎很简单，实际却需要考虑很多方面。如下面的图所示，首先，如果一个文本跟两个类别都比较相似，那它是只能属于一个类别还是能同时属于两个类别？其次，由于相似度量不满足传递性，相邻的文本比较相似，但距离比较远的文本之间其实已经不相似了，要不要把这批文本放到一个类别，还是从中间切割成两个类别？还有一个问题，我们对文本进行聚类，是直接把文本聚合到一个一个类别，还是采用层次的方法对这些类别进行组织。
接下来我们结合前面聚类的运用场景，介绍如何选择合适的相关技术的。
二、技术选择
文本的相似度量一般分为两个步骤：
1）文本向量化表征：把文本转化成高维向量空间上的点，使用高维向量来表征文本；
2）相似度计算：基于高维向量的表征形式，计算向量之间的距离。近几年来随着深度学习技术的兴起，也有使用交互式深度学习模型直接计算两个文本之间相似度的方法，不过考虑到线上未识别文本数量巨大，使用深度学习模型对它们进行两两计算基本上是不可行的。
文本向量化表征又分为两大类：基于词袋的方法和基于语义向量的方法。基于词袋的方法具体有one-hot、词频、TF-IDF、概率主题模型等方法，基于语义向量的方法则主要包括词向量、深度学习模型表征等技术。
总体而言，语义向量方法能够在语义层面上衡量文本的相似度，但错误的可解释性较差，词袋方法则相反。在我们的业务场景中，我们要处理的相似文本往往是下图的这种形式，并且对聚类结果的解释性要求比较高，因此文本向量化表征技术选择了TF-IDF方法。文本的相似度计算常用的有余弦距离、欧式距离、Jarcard相似度，由于我们选择了TF-IDF方法，相似度计算最终使用的是余弦距离。
前面讲到聚合方法需要考虑的几个问题，围绕这些问题学术界展开了相关的研究，并提出了丰富多样的聚类算法，对我们的工作具有积极的借鉴意义。不过这些方法多数需要重复迭代，在性能和实时性方面并不友好，满足不了我们对在线文本聚类的实时性需求。出于聚类实时性的要求，我们选择了过程相对简单single-pass cluster算法，该算法对新来的数据跟历史数据比较一遍相似度即可，不需要重复比较的过程，这也是算法名字的由来。
让我们注意力放在新数据跟历史数据比较这个过程，这其实就是k-近邻问题，从这个角度看，k-近邻又有大量的工作专门研究如何加速k-近邻检索过程，具体有minhash、simhash、canopy、KD-Tree以及最近比较火的Faiss开源项目。考虑到后面的分布式实现，我们选择了实现相对简单的canopy算法作为k-近邻检索加速方案。
上图是我们的文本聚类单机流程：系统接收到一个新的文本以后，首先把文本转化成高维空间上的TF-IDF向量；接下来需要从历史数据中搜索与该向量距离比较近的数据，这里先使用canopy算法进行粗聚类，从海量的历史数据中筛选出少数候选数据；然后再分别与新文本计算TF-IDF向量的余弦相似度，如果有候选数据相似度大于阈值，则把新文本归到该数据所属的类别中，否则新建一个类别，新文本作为新类别的初始成员。
上面只是线上文本实时聚类的一个原型设计，实际上的业务场景存在一些特点，对实时聚类实现方案提出更高的要求。线上的文本数据是以连续不间断数据流形式源源不断的进来的，聚类系统内部在进行聚类的同时，需要能够实时地接收每一个新的文本数据，即已接收文本的聚类处理和新文本的接收必须同时进行。实时聚类主要是服务于未识别内容审核的，当聚类中的某个类别满足我们设定的条件时，需要把该类别的所有文本发到审核系统进行人工审核，这个过程越实时越好。另外线上未识别的文本流量巨大，无论是计算量还是内存使用量，单个机器都无法承载，需要使用多台服务器共同完成实时聚类任务。
可以看出，我们的业务场景非常适合使用流计算技术，目前市面上也有比较多成熟的流计算框架，storm、flink、spark stream都是其中的佼佼者。这些流计算框架大都拥有优异的性能表现，对分布式的支持也非常友好，我们只需要基于这些框架的编程范式把上述聚类设计实现好即可。
不过这里有一个核心问题，那就是一般的流计算任务流数据之间的处理是独立的，即当前数据的处理与其他数据无关，因此比较容易实现分布式处理。而聚类问题研究的就是数据之间的关系，因此在分布式实现方面存在困难。比如为了找到当前文本的相似文本，需要对所有的历史数据作比对，这要求所有数据存放在同一个内存里面，如果分布式实现把数据放在不同服务器节点，那就很难实现这个相似查找的功能。
下面介绍一下我们是怎么解决这个问题的。
三、问题解决
如上图，分布式实时聚类系统由两个子系统组成：蓝色虚线圈起来的聚类子系统和Recorder子系统。其中Recorder负责把需要审核的数据同步到审核系统，最核心还是聚类子系统，它基于分布式流计算框架实现，包括Spout、ClusterBolt、OutputBolt 3个模块。
Spout的功能包括从kafka读数据、数据清洗分词等预处理、把文本转化成TF-IDF向量。这些功能没有涉及到多数据间的关系，因此可以实现完全的分布式。
ClusterBolt收到Spout发来的TF-IDF向量后，先进行canopy粗聚类，筛选出候选数据，再进行基于余弦聚类的聚类。由于需要历史数据，ClusterBolt无法完全分布式实现，同时存储历史数据对内存要求较高，内容容易成为系统瓶颈。考虑到聚类是发生在业务内部，不同业务之间的文本内容是不需要聚类的，我们根据业务对聚类进行划分，把不同业务的聚类分散到不同的节点上面进行，从而减少单个节点上面的数据规模。
当聚类到的类别满足审核条件时，ClusterBolt把相应的文本发送给OutputBolt，再由OutputBolt统一写到kafka。可以看到OutputBolt功能比较简单，因此分配的节点数会比较少。
接下来我们分析一下上述设计是否完成了我们的分布式目标。之所以需要分布式，无非就是单机的系统资源不够满足任务的需求（也有因为高可用需要多个服务器的情况）。对于文本实时聚类来说，主要的因素就是CPU、内存、网络带宽的需求。CPU方面，分词和余弦距离计算都是计算型的操作，系统的CPU主要消耗在这两块。内存的使用主要集中在存储原始文本数据和TF-IDF向量上面。至于带宽方面，主要是读kafka和写kafka，不过目前的情况看带宽还不构成问题。
首先，分词已经随Spout模块已分布式的方式分摊到多个节点上面了。余弦距离计算和内存使用则发生在ClusterBolt，也以业务为维度分摊到多个节点了。因此，上述设计可以认为基本达成分布式目标了。不过这里还遗留一个问题，那就是业务间的差异比较大，一方面，不同的业务数据的规模相差很大，另一方面，不同业务波峰波谷在时间上的分布也不同。因此基于这个设计的系统存在以下待优化的问题：
1）不同业务规模的不一样，导致节点之间的资源利用率不同，有些节点比较空闲，浪费系统资源；
2）由于波峰波谷的变化，节点有些时间段比较空闲，有些时间段比较繁忙，产生较大的波动性，并且不同业务的节点是隔离的，无法实现削峰平谷；
3）对于数据规模特别巨大的业务，还是存在单个节点资源不够的问题。
为了克服上述问题，我们在原有系统的基础进行改进，形成了上图所示的方案。主要改进点有两个：
1）增加了DistributeBolt模块，Spout只负责读kafka数据，数据清洗分词等预处理、把文本转化成TF-IDF向量这些转到DistributeBolt上面做。这样可以实现更精细的资源分配，把读数据和复杂计算的功能解耦，各自按需分配。
原来的ClusterBolt细分成LocalClusterBolt和MergeClusterBolt两个模块。同一个业务的数据可以分发到不同的LocalClusterBolt节点，先在节点内部进行局部聚类（聚类方法跟原来的一样），局部聚类以后符合条件的数据再发送到MergeClusterBolt进行合并。当然相同业务的数据还是需要在同一个MergeClusterBolt节点上面进行合并的。
这里可能会有一个疑问：要是运气不好同一个类别的数据刚好分发到不同局部聚类节点，比如每个节点刚好分到一条数据，那么这批数据是不是就聚不出来了？为了避免这种情况，我们对数据的分发算法进行了精心的设计，如下图所示：
一条数据只分发到一个局部聚类节点，确实存在较大的可能性聚不出来，所以我们会发数据发送给k个节点。通过对TF-IDF向量数值top k的分量对应的token id进行partition操作，得到k个局部聚类节点的编号，然后把数据同时发给这k个节点。这其实很好理解，数值比较大的分类，对余弦相似度计算的贡献越大，上述操作能保证在相同分量上面都具有较大值的文本会被分发到相同的节点。
值得一提的是，这里的k不是固定的，而是基于历史聚类数据训练机器学习回归模型，对不同的文本内容，动态决定k的值。经过实测，k的平均值大概在3左右，相似文本在同一个局部聚类节点聚出来的比例达到98%以上。但还是还会有个别数据被分发到的局部聚类节点没法聚出来，可能同个类别的数据在其他节点都已经聚出来了，只留它孤零零的在这个节点上面。我们的方案也考虑到这种情况了，下图是数据从局部聚类节点到合并聚类节点的逻辑图。
一个文本在LocalClusterBolt完成聚类流程以后，如果它所在的类别已经满足入库审核的条件，则直接把该类别所有数据发往MergeClusterBolt。MergeClusterBolt直接把收到的数据发往OutputBolt进行后续入库审核流程，同时在内存中继续保留这些数据。如果在LocalClusterBolt为满足入库审核条件，则等待一段时间后再把数据发往MergeClusterBolt。MergeClusterBolt收到数据后，在内存中的其他数据查找是否有相似的文本，如果有相似文本则把收到的数据发往OutputBolt进行后续入库审核流程，不管是否有相似数据，MergeClusterBolt永远不会保留这类数据。
四、思考
通过上面的介绍可以看出，改进后的系统已经比较好地解决了实时分布式文本聚类问题，并且随着业务规模的增长，我们的系统完全可以通过横向扩展增加服务器来实现整体处理能力的线性提升。文本聚类技术的应用对提升易盾内容安全整体防控能力起到了非常大的作用，对线上漏识别的有害内容召回效果比较突出。