上一篇我們介紹了Flink CEP的API，這一篇我們將以結合一個案例來練習使用CEP的API編寫應用程式，以強化對API的理解。所選取的案例是對網路遭受的潛在攻擊進行檢測並給出告警。當下網際網路安全形勢仍然嚴峻，網路攻擊屢見不鮮且花樣眾多，這裡我們以DDOS（分散式拒絕服務攻擊）產生的流入流量來作為遭受攻擊的判斷依據。

假定一家雲服務提供商，有多個跨地區的資料中心，每個資料中心會定時向監控中心上報其瞬時流量。

我們將檢測的結果分為三個等級：

• 正常：流量在預設的正常範圍內；
• 警告：某資料中心在10秒內連續兩次上報的流量超過認定的正常值；
• 報警：某資料中心在30秒內連續兩次匹配警告；

首先，我們構建source，這裡我們選擇的是並行source，因此需要繼承RichParallelSourceFunction類。所有的資料通過模擬器隨機生成，其中資料中心編號為整型且取值範圍為[0, 10)，資料生成的事件間隔由PAUSE常量指定，預設為100毫秒：

//parallel source
DataStream<MonitorEvent> inputEventStream = env.addSource(
    new MonitorEventSource(
        MAX_DATACENTER_ID,
        STREAM_STD,
        STREAM_MEAN,
        PAUSE
    )
).assignTimestampsAndWatermarks(new IngestionTimeExtractor<MonitorEvent>());

下面，我們來構建警告模式，按照我們設定的警告等級，其模式定義如下：

Pattern<MonitorEvent, ?> warningPattern = Pattern.<MonitorEvent>begin("first")
    .subtype(NetworkStreamEvent.class)
    .where(evt -> evt.getStream() >= STREAM_THRESHOLD)
    .next("second")
    .subtype(NetworkStreamEvent.class)
    .where(evt -> evt.getStream() >= STREAM_THRESHOLD)
    .within
 
(Time.seconds(10));

根據該模式構建模式流：

PatternStream<MonitorEvent> warningPatternStream =
    CEP.pattern(inputEventStream.keyBy("dataCenterId"), warningPattern);

在警告的模式流中篩選出配對的警告事件對，生成警告事件物件流（告警事件物件會算出，前後兩個匹配的流量事件的平均值）：

DataStream<NetworkStreamWarning> warnings = warningPatternStream.select(
    (Map<String, MonitorEvent> pattern) -> {
        NetworkStreamEvent first = (NetworkStreamEvent) pattern.get("first");
        NetworkStreamEvent second = (NetworkStreamEvent) pattern.get("second");

        return new NetworkStreamWarning(first.getDataCenterId(),
            (first.getStream() + second.getStream()) / 2);
    }
);

按照設定的等級，告警模式定義如下：

Pattern<NetworkStreamWarning, ?> alertPattern = Pattern.<NetworkStreamWarning>
    begin("first").next("second").within(Time.seconds(30));

在警告事件流中應用告警模式，得到告警模式流：

PatternStream<NetworkStreamWarning> alertPatternStream = CEP.pattern(warnings.keyBy
    ("dataCenterId"), alertPattern);

在告警模式流中匹配警告模式對，如果模式對中第一個警告物件的平均流量值小於第二個警告物件的平均流量值，則構建告警物件並輸出該物件從而形成告警流：

DataStream<NetworkStreamAlert> alerts = alertPatternStream.flatSelect(
    (Map<String, NetworkStreamWarning> pattern, Collector<NetworkStreamAlert> out) -> {
        NetworkStreamWarning first = pattern.get("first");
        NetworkStreamWarning second = pattern.get("second");

        //first avg < second avg
        if (first.getAverageStream() < second.getAverageStream()) {
            out.collect(new NetworkStreamAlert(first.getDataCenterId()));
        }
    }
);

最終，sink到控制檯：

warnings.print();
alerts.print();

從上面的程式碼段可見，CEP的關鍵是定義合適的模式。關於模式的相關的API，我們之前已進行過分析。為了節省篇幅，本文只列出了核心程式碼片段。

需要注意的是，因為包含Java 8的lambdas，當你使用javac作為編譯器時，將會得到錯誤提示：

Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException: 
The generic type parameters of 'Map' are missing. 
It seems that your compiler has not stored them into the .class file. 
Currently, only the Eclipse JDT compiler preserves the type information necessary to 
use the lambdas feature type-safely. 
See the documentation for more information about how to compile jobs containing lambda expressions.
at org.apache.flink.api.java.typeutils.TypeExtractor.validateLambdaGenericParameter(TypeExtractor.java:1331)
at org.apache.flink.api.java.typeutils.TypeExtractor.validateLambdaGenericParameters(TypeExtractor.java:1317)
at org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:347)
at org.apache.flink.cep.PatternStream.select(PatternStream.java:81)
at com.diveintoapacheflink.chapter11.NetworkAttackMonitor.main(NetworkAttackMonitor.java:55)
at ...

解決方案是使用Eclipse JDT來編譯程式碼。

微信掃碼關注公眾號：Apache_Flink

QQ掃碼關注QQ群：Apache Flink學習交流群（123414680）