摘要：所有不相關的數據會立即丟棄，由于查詢都是在一個無限的數據流中，這樣的優勢顯而易見。基于這些監控事件數據流，我們想要檢測出可能要過熱的機架，從而調整負載和降溫。

原文鏈接

隨著傳感網絡的普及，智能設備持續收集著越來越多的數據，分析近乎實時，不斷增長的數據流是一個巨大的挑戰。快速應對變化趨勢、交付最新的 BI 應用會成為一個公司成敗的關鍵因素。其中關鍵問題就是數據流的事件模型檢測。

Complex event processing (CEP) 要處理的就是在持續事件中匹配模式的問題。匹配結果通常就是：從輸入事件中提取的復雜事件。傳統 DBMSs 在固定數據上執行查詢，而 CEP 在存儲的 query 上執行（譯者注：某個范圍）。所有不相關的數據會立即丟棄，由于 CEP 查詢都是在一個無限的數據流中，這樣的優勢顯而易見。更重要的是，輸入實時被處理，系統一旦收到某一個序列的所有數據，結果就會被輸出。CEP 因此有著非常高效的實時分析能力。

由此，CEP 的處理范式吸引了很多技術人員興趣，有著廣泛的應用場景。值得注意的是，CEP 現在用在了金融應用，例如：股票市場趨勢、信用卡欺詐檢測。還有基于 RFID 的追蹤和監控，例如：庫房小偷檢測。CEP 還可以被用于基于用戶可疑行為的網絡入侵檢測。

Apache Flink 有著天生的真正的流處理能力，具有低延遲、高吞吐量的特性，和 CEP 簡直絕配。因此，Flink 社區在 Flink 1.0 引入了第一個版本的 CEP library。接下來我們會使用一個數據中心監控的案例介紹其使用。

假設這樣一個場景：數據中心有很多機架，每一個機架都有功率和溫度監控。監控設備會不斷產生功率和溫度事件。基于這些監控事件數據流，我們想要檢測出可能要過熱的機架，從而調整負載和降溫。

針對這種場景，我們采取兩階段處理方法。首先，監控溫度事件，當檢測到連續兩個超過閾值的溫度事件，即生成一個當前平均溫度的警告（warning），溫度報警不一定意味著過熱。但是如果看到兩個連續的升溫警告事件，則生成機架過熱報警（alert)。此時，需要采取措施冷卻機架。

首先，定義來源的監控事件流，每一個 message 都包含來源 rack ID（機架 ID）。溫度事件包含當前溫度，功率事件包含當前電壓。我們把事件模型定義為 POJOs.

    public abstract class MonitoringEvent {
        private int rackID;
        ...
    }
    
    public class TemperatureEvent extends MonitoringEvent {
        private double temperature;
        ...
    }
    
    public class PowerEvent extends MonitoringEvent {
        private double voltage;
        ...
    }

我們可以使用 Flink 的 connector（比如：Kafka, RabbitMQ 等），生成 DataStream inputEventStream 給 Flink 的 CEP 算子提供輸入。首先，我們需要定義檢測溫度警告的事件模式 (pattern)，CEP library 提供了非常直觀的 Pattern API 來定義復雜的模式。

每個模式都包含了一個可以定義過濾 (filter) 條件的事件序列。模式 (pattern) 的第一個事件通常都命名為"First Event"。

    Pattern.begin("First Event");

這句話會匹配每一個輸入的監控事件（monitoring event），而我們只需要溫度大于一定閾值的溫度事件（TemperatureEvents），所以我們需要添加 subtype 和 where 語句限制。

    Pattern.begin("First Event")
        .subtype(TemperatureEvent.class)
        .where(evt -> evt.getTemperature() >= TEMPERATURE_THRESHOLD);

之前說：對于同一個機架，當看到兩個連續的高溫事件（超過閾值）就產生一個溫度報警（TemperatureWarning），Pattern API 提供了 next 調用方法，來添加事件到模式定義中。next 添加的事件發生時間必須緊跟著第一個匹配事件之后，才能觸發整個模式的匹配。

Pattern warningPattern = Pattern.begin("First Event")
    .subtype(TemperatureEvent.class)
    .where(evt -> evt.getTemperature() >= TEMPERATURE_THRESHOLD)
    .next("Second Event")
    .subtype(TemperatureEvent.class)
    .where(evt -> evt.getTemperature() >= TEMPERATURE_THRESHOLD)
    .within(Time.seconds(10));

最后模式的定義包含有一個 within 的 API 調用，用來定義兩個連續 TemperatureEvents 必須在 10s 內發生才能匹配。時間基于 time characteristic 設置，可以是：處理時間、輸入時間或者事件時間。(譯者注 Event Time / Processing Time / Ingestion Time 解釋)

定義好事件模型之后，可以將其應用到輸入數據流中。

    PatternStream tempPatternStream = CEP.pattern(
        inputEventStream.keyBy("rackID"),
        warningPattern);

由于告警是針對單個機架的告警，必須使用 keyBy 通過 rackID 字段對輸入事件流分流。即匹配出的事件都是同一個機架的。

PatternStream 可以訪問匹配的事件序列。通過使用 select API 可以訪問其上數據，給 select API 傳入 PatternSelectFunction，PatternSelectFunction 會在每一個匹配上的事件序列上執行。事件序列通過 Map 訪問，MonitoringEvent 通過之前分配的事件名稱來定位。這里我們通過 select function 針對每一個匹配的模式產生一個 TemperatureWarning 事件。

    public class TemperatureWarning {
        private int rackID;
        private double averageTemperature;
        ...
    }
    
    DataStream warnings = tempPatternStream.select(
        (Map pattern) -> {
            TemperatureEvent first = (TemperatureEvent) pattern.get("First Event");
            TemperatureEvent second = (TemperatureEvent) pattern.get("Second Event");
    
            return new TemperatureWarning(
                first.getRackID(), 
                (first.getTemperature() + second.getTemperature()) / 2);
        }
    );

現在我們從原始監控事件流（monitoring event stream）生成了一個復雜事件流 DataStream 警告。這個復雜事件流可以再次被用作其他復雜事件處理的輸入。當同一個機架產生兩個連續升溫警告時，我們使用 TemperatureWarnings 來生成 TemperatureAlerts。TemperatureAlerts 定義如下：

    public class TemperatureAlert {
        private int rackID;
        ...
    }

首先定義報警事件

    Pattern alertPattern = Pattern.begin("First Event")
        .next("Second Event")
        .within(Time.seconds(20));

定義描述了在 20s 內有兩個 TemperatureWarnings 事件，并且第一個事件名稱為 "First Event"，緊接著的第二個為 “Second Event”。這來了個事件都沒有 where 語句，因為需要訪問兩個事件才能判斷溫度時候增長。因此，下面我們需要在 select 語句中使用 filter 條件來提取。這里我們只是生成了 PatternStream。

    PatternStream alertPatternStream = CEP.pattern(
        warnings.keyBy("rackID"),
        alertPattern);

同樣，我們需要 keyBy 對輸入的告警數據流針對同一個機架進行分流。然后使用 flatSelect 方法訪問匹配的事件序列，當判斷溫度上升時生成 TemperatureAlert 告警。

    DataStream alerts = alertPatternStream.flatSelect(
        (Map pattern, Collector out) -> {
            TemperatureWarning first = pattern.get("First Event");
            TemperatureWarning second = pattern.get("Second Event");
    
            if (first.getAverageTemperature() < second.getAverageTemperature()) {
                out.collect(new TemperatureAlert(first.getRackID()));
            }
        });

DataStream 告警是針對同一個機架的數據流，基于這個數據我們現在可以調整負載和降溫。源代碼地址（譯者注：注意閱讀 readme）

本文描述了使用 Flink CEP library 可以很容易處理事件流。我們通過數據中心的監控和報警案例，完成了服務器機架過熱報警的小程序。
未來 Flink 社區會持續擴展 CEP library 的功能和表述能力。接下來的 road map 是支持類正則表達式的模式實現，包括 *， 上下限制和否定。此外，還計劃允許 where 語句訪問之前匹配的事件字段。這個特性可以讓我們提前刪除不需要的事件序列。

本內容為譯者添加

官網：Apache Flink

概念：Event Time / Processing Time / Ingestion Time

案例：Apache Flink example CEP program to monitor data center temperatures

API 介紹：FlinkCEP - Complex event processing for Flink