大家好，欢迎来到IT知识分享网。

本文翻译自Flink官方blog所提供的一个基于事件驱动的Flink经典应用案例，学习本案例不仅可以对Flink在事件驱动实时数据处理场景中的应用思路，还可以领略到Flink资深专家在构建一个Flink应用时的专业编码规范，为自己的Flink程序开发提升功力

在本案例中，您将了解构建流媒体应用程序的三种强大的 Flink 模式：

• 应用逻辑的动态更新
• 动态数据分区（洗牌），在运行时控制
• 基于自定义窗口逻辑的低延迟警报（无需使用窗口 API）

这些模式扩展了静态定义的数据流可以实现的可能性，并提供了满足复杂业务要求的构建基块。

1）动态更新计算逻辑：允许 Flink 在工作运行时更改应用逻辑，而无需停止和重新提交代码。

2）动态数据分区（动态keyby）：提供了在运行时更改事件分布和数据分组的能力。在使用动态可重新配置的应用程序逻辑构建作业时，此类功能通常成为自然要求。

3）自定义窗口管理：演示了当原生窗口 API（window API） 不完全符合您的要求时，如何利用 low level process function API 来化解困难；具体来说，您将学习如何在窗口上实现低延迟警报，以及如何用定时器限制状态增长。

这些模式建立在核心 Flink 功能之上，但是，在正统的官方文档中，这些高级模式并没有进行充分说明，因为如果没有具体的用例，解释和呈现它们背后的动机会过于空洞和抽象。这就是为什么我们要用一个实用的例子（欺诈检测引擎）来展示这些特性和上述的“应用模式”，我们希望这个系列教程，将把这些强大的方法放入您的工具箱，并使您能够承担新的和令人兴奋的任务，更好地发挥Flink的强大功能。

在系列的第一篇教程中，我们将查看本案例应用（欺诈检测引擎）的高层架构，描述其组件及组件之间的交互关系。然后，我们将深入探讨该系列中第一个模式的实施细节 –动态数据分区。

您将能够在本地运行完整的欺诈检测演示应用程序，并使用随附的 GitHub 存储库查看实施的详细信息。

欺诈检测演示

我们欺诈检测演示的完整源代码是开源的，可在线获得。要在本地运行，请查看以下存储库，并按照 README 中的步骤操作：

https://github.com/afedulov/fraud-detection-demo

您将看到演示是一个自成一体的应用程序 – 它只需要和从源构建，包括以下组件：dockerdocker-compose

• Apache Kafka (message broker) with ZooKeeper
• Apache Flink (application cluster)
• Fraud Detection Web App

欺诈检测引擎的应用目标是对源源不断流入的金融交易记录，根据一系列规则进行评估。这些规则可能会被频繁地更改和调整。在真正的生产系统中，重要的是能够在运行时添加和删除这些评估规则，而不代码停止和重启job的昂贵代价。

当您按照README中的步骤，启动“欺诈检测系统”后，可以在浏览器中打开如下页面：

在左侧，您可以在单击”开始”按钮后看到流经系统的金融交易的明细记录。顶部的滑块允许您控制每秒生成的交易数量。中间部分专门管理 Flink的计算逻辑： 欺诈评估规则。从这里，您可以创建新规则以及发布控制命令，例如清除 Flink 状态。

系统中预先定义了一些示例规则。您可以单击“开始”按钮，并在一段时间后观察UI右侧显示的警报。这些警报是 Flink 根据预先定义的规则评估生成的欺诈检测结果。

我们的示例欺诈检测系统由三个主要组成部分组成：

1. 前端（React实现）
2. 后端（Springboot实现）
3. 欺诈检测应用程序（Apache Flink实现）

主要组件之间的相互作用如下图：

欺诈系统组件示意

后端将 REST API 暴露在前端，用于创建/删除规则以及发布用于管理控制的命令。然后，它通过一个kafka topic（topic_name: control） ，将这些前端操作传递给 Flink。后端还包括金融交易数据生成器组件，该组件通过 “trasaction” topic 向 Flink 发送模拟的汇款事件流。Flink 生成的警报由 “alerts” topic的后端消耗，并通过 WebSocket 传递到 前端UI。

现在，您熟悉了欺诈检测引擎的总体布局和目标，现在让我们详细了解实施此类系统所需的内容。

动态数据分区

我们将研究的第一个模式是动态数据分区。

如果您过去曾使用 过Flink 的DataStream API，则您无疑熟悉 KeyBy 方法。KeyBy算子会产生数据流的shuffle ，以便将具有相同key的元素分配到相同的分区。这意味着所有具有相同key的记录都由下一个operator的相同物理实例处理。

在典型的流式处理应用程序中，key的选择是固定的，由元素内的某些静态字段决定。例如，在构建基于窗口的简单交易流聚合时，我们可能总是按交易帐户 ID 进行分组。

DataStream<Transaction> input = // [...]
DataStream<...> windowed = input
  .keyBy(Transaction::getAccountId)
  .window(/*window specification*/);

IT知识分享网

此方法是实现各种使用案例中水平可伸缩性的主要构建基础。但是，如果应用程序在运行时尝试提供业务逻辑的灵活性，上述的KeyBy方案不够的。要了解为什么会这样，让我们首先来看看欺诈检测系统的真实规则定义示例：

“每当同一付款人向同一受益人支付的累计付款金额在一周内超过 100 万美元时 – 发出警报。

在此公式中，我们可以发现一些参数，我们希望能够在新提交的规则中指定这些参数，甚至可能稍后在运行时修改或调整：

1. 聚合字段（付款金额）
2. 分组字段（付款人+受益人）
3. 聚合功能（总和）
4. 窗口持续时间（1 周）
5. 限制阈值 （1 000 000）
6. 限制操作符（大于）

因此，我们将使用以下简单的 JSON 格式来定义上述参数：

IT知识分享网{
  "ruleId": 1,
  "ruleState": "ACTIVE",
  "groupingKeyNames": ["payerId", "beneficiaryId"],
  "aggregateFieldName": "paymentAmount",
  "aggregatorFunctionType": "SUM",
  "limitOperatorType": "GREATER",
  "limit": 1000000,
  "windowMinutes": 10080
}

此时，必须了解，groupingKeyNames必须确定事件的实际物理分组 – 所有具有相同特定参数值的交易（例如付款人#25 – >受益人#12）必须在计算评估规则的operator算子的同一物理实例中进行聚合。

keyBy()函数，在Flink文档中的大多数示例都使用硬编码，该编码提取特定的固定事件字段。但是，为了支持所需的灵活性，我们必须根据规则的规范以更灵活的方式提取它们。为此，我们将不得不使用一个额外的operator，它负责将每个事件发送到一个正确的聚合实例。keyBy() KeySelector

在上述要求上，我们的主处理管道看起来像这样：

DataStream<Alert> alerts =
    transactions
        .process(new DynamicKeyFunction())
        .keyBy(/* some key selector */);
        .process(/* actual calculations and alerting */)

我们以前已经确定，每个规则定义一个 groupingKeyNames 参数，指定哪些字段组合将用于作为事件的分组依据。每条规则都可能任意组合这些字段。同时，可能需要根据多种规则对每一个传入的事件进行评估。这意味着，事件可能需要同时出现在多个与不同规则相符的评估计算operator的并行实例中。实现这样的动态分组功能的是： DynamicKeyFunction()

DynamicKeyFunction()+keyBy()

DynamicKeyFunction在一组定义的规则（多个不同评估规则，且分组需求不同）上重复计算，并通过提取所需的分组key来交给：keyBy()函数进行正确分组

IT知识分享网public class DynamicKeyFunction
    extends ProcessFunction<Transaction, Keyed<Transaction, String, Integer>> {
   ...
  /* Simplified */
  List<Rule> rules = /* Rules that are initialized somehow.
                        Details will be discussed in a future blog post. */;

  @Override
  public void processElement(
      Transaction event,
      Context ctx,
      Collector<Keyed<Transaction, String, Integer>> out) {

      for (Rule rule :rules) {
       out.collect(
           new Keyed<>(
               event,
               KeysExtractor.getKey(rule.getGroupingKeyNames(), event),
               rule.getRuleId()));
      }
  }
  ...
}

KeysExtractor.getKey()使用反射从交易事件中提取所需的分组字段，并将它们组合为单个串联字符串key

例如，跟踪付款人#25和受益人#12之间的所有交易，并在所需的时间范围内计算预定义的欺诈检测规则，伪代码如下： groupingKeyNames”{payerId=25;beneficiaryId=12}”

请注意，示例系统的源码中，用如下POJO类，来封装分组key的信息，并传入：KeyedDynamicKeyFunction

public class Keyed<IN, KEY, ID> {
  private IN wrapped;
  private KEY key;
  private ID id;

  ...
  public KEY getKey(){
      return key;
  }
}

DataStream<Alert> alerts =
    transactions
        .process(new DynamicKeyFunction())
        .keyBy((keyed) -> keyed.getKey());
        .process(new DynamicAlertFunction())

注意，此处会因为多规则而产生隐式的事件复制

通过这样做，我们实现了一个重要属性 – 规则处理的水平可伸缩性。

我们的系统将能够通过在集群中添加更多服务器来处理更多的规则，即增加并行性。

此属性是以数据重复为代价实现的，这可能会成为一个问题，具体取决于特定的实际场景，如传入数据速率、可用网络带宽、事件有效载荷大小等。在现实实践中，可以应用额外的优化：例如对具有相同规则的规则进行综合评估，或对过滤层进行综合评估，在处理特定规则时，预先剔除不需要字段或事件。

总结：

在这篇文章中，我们通过查看示例使用案例 （ 欺诈检测引擎 ） 讨论了支持 Flink 应用程序实时、动态更改计算逻辑的动机及整体思路。我们描述了其组件之间的整体架构和交互，并为在生产实践中构建和运行演示欺诈检测应用程序提供了参考。然后，我们展示了实施动态数据分区模式的细节，作为实现运行时灵活变更计算逻辑最终目标的第一个基础模块。

为了继续专注于描述模式的核心机制，我们把数据处理和基本规则引擎的复杂性保持在最低限度。如果继续扩展，很容易想象在本架构上如何添加更高级扩展，例如允许更复杂的规则定义，包括过滤某些事件、逻辑规则链和其他更高级的功能。

在这个系列的第二部分，我们将描述“规则定义”将如何注入运行中的欺诈检测引擎。

此外，我们将详细研究管道的主要处理功能-DynamicAlertFunction( )的实施细节。

端到端管道

在下一篇文章中，我们将看到如何 Flink 的广播流在欺诈检测引擎内的应用（动态更新欺诈检测规则：即动态改变计算逻辑）。