摘要：本文整理自阿里云计算平台算法专家张颖莹，在 Flink Forward Asia 2022 AI 特征工程专场的分享。本篇内容主要分为五个部分：
阿里云大数据平台的智能运维
智能运维算法服务应用场景
传统算法工程链路的局限性
使用 Flink ML 搭建智能运维算法服务
总结和开源计划
一、阿里云大数据平台的智能运维
阿里云计算平台提供了多个非常核心的大数据计算和人工智能相关的产品，支撑了阿里集团内部以及云上各行各业客户很多核心的业务场景。在这里我挑选了三个非常典型的大数据计算产品来给大家做介绍，它们是大数据计算服务 MaxCompute、实时计算 Flink、实时数仓 Hologres。
这些产品所支撑的业务场景大家其实也都非常熟悉，比如我们日常在手机淘宝、蚂蚁森林中收取的能量数据就依赖于像 MaxCompute 大数据计算服务定时产出，因为它主要负责大规模数据的离线计算；双十一期间，我们会看到非常多炫酷的数字大屏，这些大屏上实时滚动的数字就依赖于像 Flink 实时大数据处理系统；当我们日常在手机淘宝上搜索一些商品的关键词的时候，Hologres 则会在底层帮我们进行实时的交互式分析，从而为我们推荐出实时的搜索结果。
可以看出，这几个大数据平台所支撑的业务场景是非常丰富的，它的用户规模非常的庞大，平台本身的架构也十分复杂。因此保障平台的稳定性就变成了一项重要且富有挑战性的工作。我们计算平台专门设置了一支运维中台的研发团队，也就是我所在的团队，来负责大数据平台的统一运维管控。
比较有意思的一点是，我们既是大数据平台的使用者，同时我们也负责平台的智能化运维工作。
从运维的视角来看，我们最关心的核心问题主要有三个层面，分别是稳定性、成本、效率。
在稳定性层面，涉及到的问题包括是否能够及时发现大数据平台中发生的异常、定位异常背后的根本原因、及时的止血和修复问题。
在成本层面，我们关心能否通过更加合理的资源配置以及优化应用的排布，在保障稳定性的前提下，能够让我们的成本降到最低。
在效率层面，我们不仅关注大数据平台本身的性能的提升，同时我们也希望使用大数据平台的用户能到能够得到非常高效的技术支持和答疑。
前面提到的典型场景当然都离不开数据的支撑，随着系统的云原生化以及可观测性理念的普及，我们现在所能获取到的系统层面的可观测性数据也越来越丰富了，包括指标、日志、Trace 等等多种不同形态的数据。
基于传统的人工分析已经很难实现对海量数据的全面高效分析，因此也就催生出了对于智能运维算法能力的需求。那么在智能运维场景里，我们对算法模型都有哪些需求呢？
首先，我们希望算法模型能够处理来自多个不同数据源、各个不同形态的数据。比如我们前面所提到的指标就属于时间序列类型的数据，而日志属于文本类型的数据。
同时，我们希望在智能运维场景中的算法还要具备足够高的性能。因为在运维的场景中，我们需要面对的往往是信息密度比较低的海量数据。
此外，无论是从大数据平台所支撑的业务场景来看（比如我们刚才提到的双十一的数字大屏），还是从阿里云所承诺给用户的服务质量的角度而言，我们对于智能运维场景中算法的实时性也都有非常高的要求。
二、智能运维算法服务应用场景
那么接下来我们就通过几个典型的案例来给大家介绍，在智能运维的场景中都有哪一些比较典型的算法模型，以及他们是如何应用在我们的实际业务场景中的。我们依然会从智能运维的核心场景稳定性、成本和效率三个层面进行展开。
稳定性层面，我们前面提到的关键问题是，我们能否及时发现系统中的异常。在我们实际的生产中，对应的平台的运维人员会去建立自己的一套核心指标监控大盘，但对于像阿里云这样庞大的平台，即使是核心指标，它的数量也远远超过了人工能够覆盖的范畴。因此我们就需要利用时间序列的异常检测算法，它能自动捕获在运维场景中几种比较典型的异常，包括方差的变化、均值的变化、尖峰深谷、断崖式跌落、趋势增长等等。
那么通过这些智能持续的异常检测算法，我们就能够更快的发现系统中的异常问题，最终的目的则是为了保障我们承诺给用户的稳定性 SLA。常见的稳定性 SLA 的量化标准是 MTTR，也就是问题从发生到最终解决的耗时。而时间序列异常检测算法所要解决的是其中非常关键的 MTTD 环节。因此如果我们能够去缩短时间序列异常检测算法链路侧的延时，就能够缩短 MTTD，进而缩短 MTTR，最终保障我们整体稳定性 SLA 的达成。
成本层面，我们希望通过更加合理的资源配置，使得在保障用户体验和平台稳定性的前提下，能够把成本降到最低。为了达到这个目的，需要具备以下两个基本要素。
我们需要能够对用户的负载和资源需求进行非常精准的预测。
我们在系统底层的调度侧需要具备自动扩缩容的能力。
目前随着我们整个系统的云原生化，底层系统的自动扩缩容能力已经基本具备了，因此精准的预测就变得非常关键。在过去，当我们无法对用户的负载进行精准预测的时候，通常只能根据经验设置比较恒定的值，但这种设置方法会导致低峰时段的资源浪费，高峰时段的用户资源需求又无法充分得到满足，既浪费了机器又影响了用户的体验。
如果我们有了更加精准的算法，就能更加精准的捕获用户负载的复杂的周期性，同时可以根据精准的预测结果，分不同的时段设置更加合理的资源配置，使得在低峰时候资源的浪费减少，高峰时候用户的资源需求又能尽可能的得到保障。
当然我们知道在实际的业务场景中，用户的资源需求不是完美且平稳的周期性曲线。因此我们需要结合最新的数据，及时对我们的预测算法做出调整，调整我们的预测结果以及相应的置信度。最终优化我们给用户推荐的资源配置策略，或者修正前期错误的一些资源配置策略。因此我们对于时间序列预测算法链路上延时的改造，可以使得我们能够更加精准灵活的去应对线上资源的突变情况。
效率层面，我们的大数据计算平台上运行着非常多的用户的作业，无论是系统层面，还是用户自身操作层面的问题，都有可能导致作业出现报错。当用户接收到系统报错的时候，大多数用户都不知道如何处理。因此他们通常的做法就是，通过提工单的方式来寻求专业技术支持的帮助。
而随着我们用户体量的不断扩大，技术支持的答疑工作量也变得越来越大，而且其实很多的报错都是同一个原因造成的，所以给技术支持带来了非常多低效重复的答疑。因此我们希望利用日志聚类算法来实现对海量原始日志的高效压缩，使我们能够把海量的原始日志聚合成数量非常有限的日志类别。这样运维人员只需针对研发数量非常有限的日志类别，再结合他们的专家经验标注对应的解决方案即可。
当下一次其他用户遇到类似问题的时候，我们就能通过日志聚类算法去匹配报错所对应的解决方案，并通过答疑机器人直接传递给用户。从而大大提升工单的拦截率，减少技术支持的答疑工作量。同时对于用户来说，也能在接收到报错很短的时间内，就得到解决方案的推荐，进而提升用户侧的满意度。
那么在这里我们为什么会需要用到流式的日志聚类呢？
首先，我们的系统日志是源源不断生成的。其次，以往离线进行训练的算法链路，会导致用户在接收到报错，来寻求技术支持帮助的时候，底层的日志都还没聚类完成，因此也就无法推荐出合适的解决方案了。所以利用流式日志聚类算法，我们能够保证用户在接收到报错很短的时间内，就能够完成聚类，并且推荐出合适的解决方案。
通过这三个场景的讲解，我们可以了解到在智能运维场景中，这几个算法模型的实时性都有非常高的要求，因此传统的算法链路通常难以满足我们的需要。
三、传统算法工程链路的局限性
接下来我们就通过具体的传统算法工程链路来给大家讲解它的局限性。上图展示了我们以往在实践中经常会使用到的传统算法工程链路。
在数据的输入层，我们有一个流式产生的数据源。比如它可能是各个模块产生的日志数据，或者是各个模块不断上报上来的指标数据。这些数据的格式通常是比较杂乱的，所以我们需要利用 Flink 作业对数据进行实时的预处理。把它加工成我们后续算法所期望的格式，并把它输出、存储到一个数据库中。
而在传统的链路中，我们的 AI 算法部分通常是部署在单机上的一段代码，并通过调度机制进行定时的调度。这段算法会离线的从数据库中读取一批数据，进行模型的离线训练，并把训练好的模型也存储下来。
在我们实际的运维实场景中，通常会需要一些比较实时性的分析结果，所以我们以往都是去增加另外一个 Flink 作业。在这个作业中，一方面我们会去对接源头的流式数据源，同时我们也会去调用离线训练好的模型。这样的话就能使得我们的分析结果的实时性，不会受到模型本身训练频率的影响。
这样的做法虽然勉强能够运转，但它仍然存在着非常大的局限性。首先我们可以看到，整个流程非常冗长。数据的预处理阶段、模型的共建阶段、后续的数据分析阶段都分散在了各个不同链路中。我们涉及到了两个作业和一段代码脚本，同时上一个阶段的产出是被下一个阶段强依赖的，因此整个链路的运维保障成本非常高。
同时，我们刚才提到了 AI 算法部分是通过部署在机器上的定时调度的程序去执行的，因此它的实时性非常低。此外，我由于部署在单机上，算法的性能也很难得到保证，所以我们需要去对这一套传统算法工程链路进行改造优化。
四、使用 Flink ML 搭建智能运维算法服务
那么为什么 Flink ML 会成为我们的最佳选择呢？在回答这个问题之前，我们首先需要分析一下在智能运维场景中，这些经典的算法模型都具备哪些特点。
我们在前面提到了，运维中的三大核心问题，稳定性、成本、效率。这三个领域中都分别存在一些各自比较经典的算法模型。比如在稳定性层面，为了更好的去检测指标中的异常，我们会用到时间序列异常检测算法；在成本层面，为了能够更好的去提前进行资源的优化配置，我们会用到时间序列预测算法；在效率层面，为了构建日志知识库，我们会用到日志聚类算法。
这三个算法模型虽然解决的具体问题不同，但它们都具备一些共同特性。
对于时间序列相关的两个模型，比如异常检测和预测算法，他们所需要的训练数据相对较小，因此它们的模型训练速度相对来说比较快。而对于日志聚类算法，它本身的算法复杂度非常高，但是我们通过一些设计便可以实现数据的增量训练。
这些模型对于新数据比较敏感。具体的来说，新数据可能会影响到持续异常检测对于“什么是正常模式”的判断，新数据也会影响到时间序列算法对趋势的判别。此外，新数据还会影响到日志聚类算法对于聚类中心的计算。因此这三种算法模型都需要结合最新数据及时对模型进行更新。
在智能运维场景中，我们对模型本身的可解释性要求比较高，所以在实际的应用中，我们都倾向于使用一些传统的机器学习模型。
通过分析这几个经典算法模型的共同特点，接下来我们就开始思考，我们应该采取什么样的方式，来对我们的算法架构进行优化升级。我们发现 Flink ML 具备的一些特性，不仅和我们前面所分析的几个算法模型的特点非常匹配，而且它也能解决我们前面所提到的几个痛点问题。
Flink ML 是一套面向传统实时机器学习的框架，它开始于 2021 年 4 月，是 Apache Flink 的子项目，因此我们使用 Flink ML 就能够充分的利用到 Flink 本身的一些特性。首当其冲的当然是实时性，其次是 Flink 流批一体的特性。所谓流批一体是指，Flink 同时支持在无线数据流上的流式处理，以及有线数据流上的批量处理。这样的话就能够满足我们在模型训练过程中不同形态的需求。
此外，Flink 还提供了 CDC 连接器来实现数据的增量读取，所谓 CDC 就是 Change Data Capture，它可以实时读取数据库中的存量历史数据，并监控和捕获在这期间数据库的变动，从而实现数据的增量更新。CDC 的特性使得我们可以在很大程度上，降低对于数据库的读写压力，减少网络开销，减少整个链路的延时。
除了前面提到的这三个 Flink 相关的特性以外，Flink ML 本身还具备另外两个特性。首先它提供对于用户非常友好的 API，因此在很大程度上减少了用户使用和理解框架的成本。用户不需要了解算子背后的具体实现细节，就可以直接调用，因此可以提升开发效率。
此外，还有一点很重要的特性是，Flink ML 提供了非常完备的机器学习基础设施。从特征工程到后续的具体的分类聚类回归模型，Flink ML 都提供了非常丰富的机器学习相关的算子，因此很大程度上也提供了 Flink ML 的可复用性。
正是因为 Flink ML 具备的优良特性，我们决定使用它来对我们的智能运维相关的算法服务的架构进行升级。
接下来我们就以日志聚类算法为例，为大家展示具体的升级改造过程。
在进入具体的流程前，我们先来了解一下日志聚类的业务背景和面临的挑战。我们前面已经提到过，在成本层面，我们会利用日志聚类算法来实现对海量日志的高效聚合，帮助运维人员结合他们的专家经验来构建出日志知识库，从而提升后续运维环节的效率。但是要实现这样的目的并不是件容易的事，因为我们的系统日志，特别是报错日志具有以下的几个特点和挑战：
- 我们的系统日志是海量，且信息密度非常低，这就对我们的算法性能有非常高的要求。
- 我们的日志数据是文本类型的，也就是非结构化的数据，它相对于结构化的数据，处理上要复杂的多。
- 我们的日志中包含了非常多的变量，特别是在我们大数据领域的日志，通常包含非常多的表名、列名等等变量。
- 这些日志一方面具有一定的格式规范，同时它也具备一定的语义性。因为我们整个系统的规模非常庞大，开发人员也非常的众多。
- 我们的系统日志是实时生成的，因此随时都有可能出现新的日志类型，所以传统的那些基于正则表达式或者是规则对日志的进行监控或者是知识库的建设方案都不太可行。
所以对于我们的日志聚类算法而言，它需要能够自动实时的将海量的数据聚合为有限的日志类别，同时需要考虑日志中的语义性，并且还需要能够应对日志中可能随时出现的新的日志类型。
接下来我们通过具体的事例，让大家更直观的了解到，日志聚类究竟在完成一件什么样的事情。
从上图中大家可以看到六段日志原文，其中前三段日志原文除了表名不同以外，日志的其他部分都完全一致。后三条日志虽然在表述方式上与前面的三条不尽相同，但从语义上它们想表达的意思是相同的。因此在系统层面，我们认为这六条日志原文应该都属于同一个日志类型，所以我们希望，通过日志聚类算法来把这六条日志原文聚合成同一个日志类别。
那么我们怎么实现这样的目的呢？首先在预处理阶段，我们会剔除日志原文中的变量，因为这些变量对后续的聚类并没有太大的帮助。剔除变量之后，我们就能够得到日志模板，而日志的模板的长度是无法控制的。有些日志的原文可能非常长，因此它的日志模板也非常长。所以我们需要利用 hash 编码的方式，来对日志模板进行标识，方便后续的分析。
通过预处理以后，我们就能够把前面的六条日志原文，转换为这里的两条日志模板。接下来我们就需要去对这两条日志模板，进行语义层面的进一步分析。
首先我们会对这两条日志模板进行分词，并选择一些信息量比较大的词作为我们的特征列表。接下来就可以基于这些特征列表，对我们的日志模板进行向量化的构建。
具体的构建过程是，当这个词出现在日志模板中的时候，我们就会把相应的位置置为 1，否则置为零。这样我们就能够得到对应日志模板的向量表示，有了这样一个特征矩阵以后，我们就可以继续调用聚类算法来实现对日志模板的聚合了。
可以看到，这两个日志模板的大多数特征都是完全一致的，所以我们在计算向量之间的距离的时候，这两个日志模板的距离会非常的接近。所以很大概率，它能通过后续的聚类算法聚合到同类别中，接下来我们会根据我们的特征列表以及每个类别中各个日志模板包含的特征来提取出关键词列表，增强我们聚类结果的可解释性。
同时我们的研发人员也可以针对这类日志类别进行对应解决方案的标注，这样的话，它就不需要去对原始的六条日志分别进行标注，只需要对整个类别进行统一的解决方案标注，极大提升了构建日志知识库的效率。
上面的过程在我们的算法中是怎样的流程呢？如上图所示，首先我们从 SLS 中读取日志数据，然后对读取过来的日志数据进行预处理和编码，得到日志模板。并继续进行后续的文本层面的分析，包括分词、特征选择、提取关键词、日志模板的特征表示、标准化。
接着我们会调用层次聚类算法来得到我们的聚类结果，也就是各个日志模板所对应的日志类别，并把它写到数据库中去。当新的日志模板生成的时候，我们会从数据库中去读取已有的聚类结果，并从各个日志类别中选择一些具有代表性的日志模板，和我们新产生的那些日志模板一起进行新一轮的聚类，这样就实现了数据的增量训练。
那么这样一个算法流程是怎么在 Flink ML 的架构上得以实现的？
首先我们会建立一个 Flink 作业，在这个 Flink 作业中，我们依然会去 SLS 中读取流式生成的数据，而接下来的环节，因为我们需要将新生成的数据与数据库中的存量聚类结果进行统一分析，因此我们会在 Flink 作业中开发一个 Python UDF。在这个 UDF 中我们会去实现 SLS 流式数据和 RDS 存量数据的拼接，同时在这个环节，我们也会使用到前面提到的 Flink CDC 连接器来直线数据的增量读取。
接下来的环节，与我们传统的算法链路中比较一致，我们会去开发一个 Python UDF 来实现数据的预处理和编码，得到日志模板。接下来的几个环节在我们以往的链路中，其实是通过部署在机器上的 Python 脚本实现的。现在我们也都把它们迁移到了 Flink 作业中来，具体包括分词、特征选择、关键词的提取以及日志模板的向量化和标准化。目前这三个环节都还是通过 Python UDF 实现的，它所对应的 Flink ML 算子也正在紧锣密鼓的开发中，后续我们也会把这几个环节都替换为对应的 Flink ML 算子。
在完成了特征向量的构建以后，接下来的一步，我们调用的是 Flink ML 的算子来实现层次聚类。在得到聚类结果以后，我们会去更新各个日志类别的代表性日志，也就是更新我们各个聚类结果的聚类中心，并把新的数据写回到 RDS 中。这个时候 RDS 中数据的更新就会被我们前面所提到的 Flink CDC 连接器感知到，从而实现数据的增量读取。
虽然我们的整个 Flink ML 框架是针对日志聚类算法设计的，但其中的很多环节都是可以复用的。具体的来说，在数据的读取阶段，无论是从 SLS 中读取流式数据，还是从数据库中去读取存量的数据，我们都可以使用 Flink 本身的一些算子进行复用。
在特征工程阶段，虽然我们目前还使用 PythonPDF，但后续这一部分全都可以替换成对应的 Flink ML 算子。具体的来说，在分词和向量化阶段我们可以使用 CountVectorizer 算子；使用 TF-IDF 算子进行特征选择；在特征标准化阶段，可以去调用 StandardScaler 算子。
接下来的环节，对于层次聚类，我们目前调用的 Flink ML 算子也是完全开源的，大家如果有类似的需求可以直接进行调用。同时 Flink ML 层次聚类算子还提供了非常多用户可以自定义的参数，比如用户可以修改向量之间距离的计算方式等等。
此外，除了层次聚类算子以外，Flink ML 还提供了其他的一些聚类算法，比如 K-MEANS 等等。在这些算子参数中，有一个参数非常关键，它也是使 Flink ML 算子区别于其他像 Spark ML 算子的重要参数，那就是 Windows 窗口参数。
也正是因为窗口参数的存在，使得我们在无限数据流上的实时聚类成为可能。所谓窗口参数，它就是把无限的数据流聚合成为比较有限的 Mini-batch 形式的数据。
Flink ML 也提供了多种窗口策略可供选择，我们可以根据系统时间进行窗口的计算，也可以根据记录的时间进行窗口的计算，也可以按照实际流入的记录数进行窗口的计算。但无论我们选择哪一种窗口策略，窗口与窗口之间都是不会重叠的，这样我们就能防止脏数据的生成。而窗口大小的设置大家则可以根据自己对于实时性的需求进行灵活的调节。
到这里我们就可以来总结一下，我们使用 Flink ML 进行算法链路升级以后究竟有哪些收益。为了更好的进行对比，我们首先来看一下旧链路的架构。
在旧链路中，我们会使用 Flink 作业从 SLS 中读取数据，并进行预处理和编码。预处理和编码之后的数据，我们会写入到另外一个 SLS 中，新的 SLS 会存储我们的日志原文，它对应的日志模板以及编码，还有一些原始日志中就包含的一些业务维度相关的信息。
在传统链路中，我们的 Python 脚本会定期从中批量的获取新的数据，并且从 RDS 中获取之前的聚类结果，对这两部分数据进行合并，构建特征工程，完成聚类，并将结果写回到 RDS 中。而在旧链路中，我们还有一个分析的链路，即我们通常需要去对 SLS 的数据和 RDS 的数据进行联合分析。比如我们可能需要去统计每一个日志类别所对应的底层的日志的数量，或者可能需要根据一段日志原文查询它所对应的日志类别是什么。
由于这两部分的数据是分别由 Flink 作业和 Python 脚本生成的，它们也分别存储在 SLS 和 RDS 中。而 SLS 的数据是流式的，我们无法对原有的数据进行修改，因此在分析阶段，我们不得不对 SLS 和 RDS 进行频繁的联合查询，才能够得到最终的分析结果，并且以 API 的方式提供给我们的用户或者是外部的系统。
那么经过 Flink ML 改造以后的新链路是什么样的？首先我们可以看到，之前部署在机器上，定时进行调度的 Python 脚本已经不见了。之前在 Python 脚本中执行的特征工程和聚类部分已经完全迁移到了前面的 Flink 作业中。同时由于我们的预处理，特征工程和聚类都在同一个 Flink 作业中进行，因此在结果的输出阶段，原本我们只存储在 RDS 中的聚类结果，也能同时作为新的字段增加到 SLA 中进行写入。
这样在后续的分析环节，就不再需要非常高频的对 SLS 和 RDS 进行联合查询，同时也降低了我们的分析成本。所以到这里我们可以来总结一下，我们使用 Flink ML 进行链路升级的收益。
首先从链路的整体延时来看，过去因为考虑到 Python 脚本的性能，以及定时调度的问题，我们总体的链路延时会在五分钟左右。而现在我们基于 Flink ML 进行链路升级以后，整体的延时可以达到 30 秒。
此外，大家也可以从流程图中直观的看到，我们进行了链路改造以后，整个流程变得简洁了很多，我们不再需要单独维护 Flink 作业和 Python 脚本，只需要保障一个 Flink 作业的正常运行，因此在运维成本方面能够得到显著的降低。
此外，我们刚才也提到，基于新的链路，我们不再需要高频的去对 SLS 和 RDS 进行联合分析，因此我们的分析成本也得到了很大程度的降低。
最后，在算法本身的性能层面，之前的算法是部署在单机上的，它的性能无法保障。而现在我们把算法替换成了 Flink ML 的算子，因此它在性能层面可以得到更多的提升和保障。
五、总结和开源计划
接下来让我们简单的对今天的分享做一下回顾和总结。
我们的分享从运维场景中三大核心问题，稳定性、成本、效率出发，并为大家介绍了这三个场景中非常典型的三种算法模型，分别是时序异常检测、时序预测、日志聚类，并且我们通过实际的案例来给大家分享了，算法模型是如何在我们的实际业务场景中发挥效用的。
同时从这些实际的业务场景中，我们也可以更加直观的感受到，机器学习模型对于实时性和性能层面都有着非常高的要求。而 Flink ML 所具备的特性刚好能够完美解决我们的痛点，它具备实时性、流批一体、CDC 增量读取等 Flink 本身的特性。此外 Flink ML 本身还提供了非常易用的 API 和非常丰富的机器学习基础设施。
在今天的分享中，我们也以日志聚类算法为例，为大家讲解了我们整体的链路改造过程。并且从链路的延时、成本、性能层面给大家介绍了我们的整体收益。
最后我也给大家预告一下，我们基于 Flink ML 算法服务的开源计划。
SREWorks 是我们团队开源的云原生数字运维平台，目前在 github 上的 star 数已经超过了 1200 个。SREWorks 中有一个很重要的模块就是智能运维算法服务模块，我们会把我们的算法服务中具备足够通用性的算法抽取出来，在未来进行开源。其中也包括一些我们基于 Flink ML 搭建的算法服务，欢迎同行朋友们关注我们的 SREWorks，也非常期待有机会可以跟大家合作共建更好的大数据生态。
