基于 Pulsar Functions 的事件处理设计模式

本文将介绍一些常见的实时流式传输模式及其实现。

模式 1：动态路由

首先回顾一下如何使用 Apache Pulsar Functions 实现基于内容的路由。基于内容的路由是一种集成模式。该模式已经存在多年，通常用于事件中心和消息框架中。基本思路是检查每条消息的内容，根据消息内容将消息路由到不同目的地。

下面的例子使用了Apache Pulsar SDK，SDK 允许用户配置三个不同的值:

• 用于在消息中查找匹配的正则表达式

• 消息匹配表达式模式时被发送到的 topic

• 消息不匹配表达式模式时被发送到的 topic

这个例子证明了 Pulsar Functions 功能强大，可以基于功能逻辑动态决定将事件发送到哪里。

 import java.util.regex.*;
  import org.apache.pulsar.functions.api.Context;
  import org.apache.pulsar.functions.api.Function;


  public ContentBasedRoutingFunction implements Function<String, String> {


     String process(String input, Context context) throws Exception {
         String regex = context
             .getUserConfigValue(“regex”).toString();
         String matchedTopic = context
             .getUserConfigValue(“matched-topic”).toString();
         String unmatchedTopic = context
             .getUserConfigValue(“unmatched-topic”).toString();


         Pattern p = Pattern.compile(regex);
         Matcher m = p.matcher(input);
         if (m.matches()) {
           context.publish(matchedTopic, input);
         } else {
           context.publish(unmatchedTopic, input);
         }
     }
  }

模式 2：过滤

如果想通过仅保留满足给定条件的事件来排除 topic 上的大多数事件时，应用选择过滤模式。过滤模式对仅查找感兴趣的事件特别有效，如信用卡付款超过了一定金额；日志文件中的 ERROR 消息；传感器读数超过特定阈值等等（见模式 4）。

假如用户在监视信用卡交易的事件流，并尝试检测欺诈或可疑行为。由于交易量很大，选择“同意/不同意”的时间有限，用户必须先过滤掉有“风险”特征的交易，如预付现金、大额付款等。

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import com.company.creditcard.Purchase;


public class FraudFilter implements Function<Purchase, Purchase> {


    Purchase process(Purchase p, Context context) throws Exception {
         if (p.getTransactionType() == ‘CASH ADVANCE’) ||
             p.getAmount > 500.00) {
            return p;
         }
        return null;
    }
}

可以使用过滤器来过滤有“风险”特征的交易。过滤器可以识别这些“风险”特征，并只将这些交易路由到一个单独的 topic 上以进行进一步评估。

经过过滤器过滤后，所有信用卡支付都可以被路由到一个“潜在欺诈行为”的 topic 上进行进一步评估，而其他事件则会被过滤掉，过滤器也不会对过滤掉的事件执行任何操作。

上图是基于三个独立支付对象的 FraudFilter function。第一次支付符合给定标准，被路由到“潜在欺诈行为”topic 上进行进一步评估；而第二次和第三次支付不符合欺诈标准，直接被过滤掉（没有被路由到“潜在欺诈行为”过滤器上）。

模式 3：转换

转换模式用于将事件从一种类型转换为另一种类型，或用于添加、删除或修改输入事件的值。

|| 投影

投影模式类似于关系代数中的投影算子，选择输入事件的属性子集，并创建仅包含这些属性的输出事件。投影模式可用于删除事件中的敏感字段，或者只保留事件中的必要属性。下图为投影模式的一种应用，在将记录发布到下游 topic 前，“屏蔽”传入的社安全号码。

|| 富集模式

富集模式用于将数据添加到输入属性中不存在的输出事件中。典型的富集模式包含基于输入事件中的某个键值对引用数据进行某种查找。以下示例展示了如何根据输入事件中包含的 IP 地址将地理位置添加到输出事件。

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;
import com.company.creditcard.Purchase;
import com.company.services.GeoService;


public class IPLookup implements Function<Purchase, Purchase> {
    Purchase process(Purchase p) throws Exception {
      Geo g = GeoService.getByIp(p.getIPAddress());
      // By default, these fields are blank, so we just modify the object
      p.setLongitude(g.getLon());
      p.setLatitiude(g.getLat());
      return p;
    }
}

|| 分离模式

在分离模式下，事件处理器接收单个输入事件，并将其分为多个输出事件。当输入事件是一个包含多个单独事件（如日志文件中的 entry）的批处理，并且想要单独处理每个事件时，分离模式十分适用。下图展示了分离模式的处理过程：先根据换行符分隔输入，再逐行发布到配置的输出 topic。

此 function 的实现过程如下：

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;


public class Splitter implements Function<String, String> {


    String process(String s, Context context) throws Exception {
       Arrays.asLists(s.split(“\\R”).forEach(line ->
            context.publish(context.getOutputTopic(), line));
       return null;
    }
}

模式 4：警报和阈值

警报和阈值模式可进行检测，并根据检测条件生成警报（如高温警报）。可以基于简单的值，也可以基于较复杂的条件（如增长率、数量的持续变化等）生成警报。

下面的示例为基于用户配置的阈值参数（如 100.00，38.7 等）和接收警报通知的邮箱地址生成警报。当此 function 接收到超过配置阈值的传感器事件时，将发送电子邮件。

import javax.mail.*;
import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;


public SimpleAlertFunction implements Function<Sensor, Void> {
   Void process(Sensor sensor, Context context) throws Exception {
       Double threshold = context
           .getUserConfigValue(“threshold”).toString();
       String alertEmail = context
           .getUserConfigValue(“alert-email”).toString();


      if (sensor.getReading() >= threshold) {
         Session s = Session.getDefaultInstance();
         MimeMessage msg = new MineMessage(s);
         msg.setText(“Alert for Sensor:” + sensor.getId());
         Transport.send(msg);
      }
      return null;
  }
}

下面是一个有状态 function 示例，该 function 根据特定传感器读数的增长率生成警报。在决定是否生成警报时，需要访问以前的传感器读数。

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;


public ComplexAlertFunction implements Function<Sensor, Void> {


  Void process(Sensor sensor, Context context) throws Exception {
      Double threshold = context
           .getUserConfigValue(“threshold”).toString();
      String alertTopic = context
           .getUserConfigValue(“alert-topic”).toString();


      // Get previous & current metric values
      Float previous = context.getState(sensor.getId() + “-metric”);
      Long previous_time = context.getState(sensor.getId() + “-metric-time”);
      Float current = sensor.getMetric();
      Long current_time = sensor.getMetricTime();


      // Calculate Rate of change & compare to threshold.
      Double rateOfChange = (current-previous) /
                           (current_time-previous_time);
      if (abs(rateOfChange) >= threshold) {
         // Publish the sensor ID to the alert topic for handling
         context.publish(alertTopic, sensor.getId());
      }


      // Update metric values
      context.putState(sensor.getId() + “-metric”, current);
      context.putState(sensor.getId() + “-metric-time”, current_time);
  }
}

通过 Apache Pulsar Functions 状态管理特性仅保留先前的度量读数和时间，并将传感器 ID 添加到这些值中（因为将会处理来自多个传感器的度量，所以需要传感器 ID）。为了简单起见，假设事件以正确的顺序到达，即始终是最新读数，没有乱序读数。

另外，这一次我们将传感器 ID 转发到一个专门的警报 topic 以进行进一步处理，而不是仅发送电子邮件。通过这种方式，我们可以对事件进行额外的富集处理（通过 Pulsar Functions）。例如，查找获取传感器的地理位置，然后通知相关人员。

模式 5：简单计数和窗口计数

简单计数和窗口计数模式使用了聚合函数，聚合函数将事件的集合作为输入，并通过对输入事件应用一个 function 生成一个所需的输出事件。聚合函数包括：求和、平均值、最大值、最小值、百分位数等。

以下为使用 Pulsar Functions 实现“字数统计”的示例，计算每个单词在给定 topic 中出现次数的总和。

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;


public WordCountFunction implements Function<String, Void> {


   Void process(String s, Context context) throws Exception {
      Arrays.asLists(s.split(“\\.”).forEach(word -> context.incrCounter(word, 1));
      return null;
   }
}

考虑到流数据 source 无休止的特性，无限期聚合用处不大，因为通常是在数据窗口上进行这些计算（如前一小时内的故障次数）。

数据窗口代表事件流的有限子集，如上图所示。但是，应该如何定义数据窗口的边界？有两个用于定义窗口的常用属性：

• 触发策略：控制执行或触发 function 代码的时间。Apache Pulsar Function 框架通过这些规则来通知代码处理窗口中收集的全部数据。

• 清除策略：控制保留在窗口中的数据量。这些规则用于决定是否从窗口中清除数据元素。

这两个策略都是由时间或窗口中的数据量驱动的。二者之间的区别是什么？又是怎样协同工作的？在多种窗口技术中，最常用的是滚动窗口和滑动窗口。

|| 滚动窗口

窗口已满是滚动窗口清除策略的唯一条件，因此，只需要指定想要使用触发策略（基于计数或基于时间）即可。基于计数的滚动窗口是怎样工作的？

在下图的第一个示例中，触发策略设置为 2，也就是说，在窗口中有两个项目时，触发器将会触发，开始执行 Pulsar Function 代码。这一系列行为与时间无关，窗口计数达到 2 用了 5 秒还是 5 个小时并不重要，重要的是窗口计数达到 2。

将上述基于计数的滚动窗口与基于时间的滚动窗口（时间设置为 10 秒）进行对比。经过 10 秒的间隔后，无论窗口中有多少事件，function 代码都会被触发。在下图中，第一个窗口中有 7 个事件，而第二个窗口中只有 3 个事件。

|| 滑动窗口

滑动窗口计数定义了窗口的长度，窗口长度设置了清除策略以限制保留待处理的数据量；滑动间隔定义了触发策略。滚动窗口策略和滑动窗口策略都可以根据时间（时间段）或长度（数据元素的数量）来定义。

在下图中，窗口长度为 2 秒，也就是说，2 秒以前的数据会被清除，并且不会用于计算。滑动间隔为 1 秒，即每 1 秒钟执行一次 Pulsar function 代码。这样，可以在整个窗口长度内处理数据。

前面的示例都是基于时间来定义清除策略和触发策略，也可以根据长度来定义清除策略或触发策略，或者同时定义这两种策略。

在 Pulsar Functions 中实现这两种类型的窗口 function 都很容易，只需要指定一个 java.util.Collection 作为输入类型，如下所示，并在创建 function 时在 -userConfig 标志中指定适当的窗口配置属性。

用于实现前面提到的时间窗口四种情形的配置参数如下：

• “–windowLengthCount”：每个窗口的消息数量

• “–windowLengthDurationMs”：窗口时间（以毫秒为单位）

• “–slidingIntervalCount”：窗口滑动后的消息数量

• “–slidingIntervalDurationMs”：窗口滑动后的时间

正确的组合方式如下表：

时间，滑动窗口	-windowLengthDurationMs = XXXX -slidingIntervalDurationMs = XXXX
时间，Batch Window（即滚动窗口)	-windowLengthDurationMs = XXXX
长度，滑动窗口	-windowLengthCount = XXXX -slidingIntervalCount = XXXX
长度，Batch Window（即滚动窗口）	-windowLengthCount = XXXX

---

总结

本文介绍了几种使用 Apache Pulsar Functions 实现通用流处理模式的方式。这些处理模式包括基于内容的路由、过滤、转换、警报、简单计数应用程序等，还介绍了基本的窗口概念，以及 Apache Pulsar Functions 提供的窗口功能等。