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

原文鏈接

隨著傳感網(wǎng)絡(luò)的普及，智能設(shè)備持續(xù)收集著越來越多的數(shù)據(jù)，分析近乎實時，不斷增長的數(shù)據(jù)流是一個巨大的挑戰(zhàn)?？焖賾?yīng)對變化趨勢、交付最新的 BI 應(yīng)用會成為一個公司成敗的關(guān)鍵因素。其中關(guān)鍵問題就是數(shù)據(jù)流的事件模型檢測。

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

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

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

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

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

首先，定義來源的監(jiān)控事件流，每一個 message 都包含來源 rack ID（機架 ID）。溫度事件包含當(dāng)前溫度，功率事件包含當(dāng)前電壓。我們把事件模型定義為 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 來定義復(fù)雜的模式。

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

    Pattern.begin("First Event");

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

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

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

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 調(diào)用，用來定義兩個連續(xù) TemperatureEvents 必須在 10s 內(nèi)發(fā)生才能匹配。時間基于 time characteristic 設(shè)置，可以是：處理時間、輸入時間或者事件時間。(譯者注 Event Time / Processing Time / Ingestion Time 解釋)

定義好事件模型之后，可以將其應(yīng)用到輸入數(shù)據(jù)流中。

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

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

PatternStream 可以訪問匹配的事件序列。通過使用 select API 可以訪問其上數(shù)據(jù)，給 select API 傳入 PatternSelectFunction，PatternSelectFunction 會在每一個匹配上的事件序列上執(zhí)行。事件序列通過 Map 訪問，MonitoringEvent 通過之前分配的事件名稱來定位。這里我們通過 select function 針對每一個匹配的模式產(chǎn)生一個 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);
        }
    );

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

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

首先定義報警事件

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

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

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

同樣，我們需要 keyBy 對輸入的告警數(shù)據(jù)流針對同一個機架進行分流。然后使用 flatSelect 方法訪問匹配的事件序列，當(dāng)判斷溫度上升時生成 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 告警是針對同一個機架的數(shù)據(jù)流，基于這個數(shù)據(jù)我們現(xiàn)在可以調(diào)整負載和降溫。源代碼地址（譯者注：注意閱讀 readme）

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

本內(nèi)容為譯者添加

官網(wǎng)：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