开发者社区> nicenelly> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

Flink之CEP案例分析-网络攻击检测

简介: 上一篇我们介绍了Flink CEP的API,这一篇我们将以结合一个案例来练习使用CEP的API编写应用程序,以强化对API的理解。所选取的案例是对网络遭受的潜在攻击进行检测并给出告警。当下互联网安全形势仍然严峻,网络攻击屡见不鲜且花样众多,这里我们以DDOS(分布式拒绝服务攻击)产生的流入流量来作为遭受攻击的判断依据。
+关注继续查看

上一篇我们介绍了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来编译代码。


原文发布时间为:2017-03-01
本文作者:vinoYang
本文来自云栖社区合作伙伴CSDN博客,了解相关信息可以关注CSDN博客。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
【建议收藏】Flink watermark分析实战(下)
【建议收藏】Flink watermark分析实战
0 0
【建议收藏】Flink watermark分析实战(上)
【建议收藏】Flink watermark分析实战
0 0
基于 Flink x TiDB,智慧芽打造实时分析新方案
智慧芽数据仓库架构师曲明星在 FFA 2021 的演讲。
0 0
Flink数据源拆解分析(WikipediaEditsSource)
WikipediaEditsSource类作为数据源负责向Flink提供实时消息,今天咱们一起来分析其源码,了解Flink是怎么获取到来自远端的实时数据的
0 0
【Flink】(十三)Flink CEP Library 使用案例分析
【Flink】(十三)Flink CEP Library 使用案例分析
0 0
通过Flink+NBI可视化构建实时分析系统
Flink: Apache Flink是一个计算框架和分布式处理引擎,用于对无界和有界数据流进行有状态计算。其针对数据流的分布式计算提供了数据分布、数据通信以及容错机制等功能。 Flink主要特点: 1、高吞吐、低延迟、纯流式架构; 2、支持对乱序事件的处理; 3、有状态、提供exactly-once计算; 4、高度灵活的窗口机制; 5、失败恢复、故障转移、水平扩展; 6、批处理、流处理统一的API
0 0
【Flink】Flink作业调度流程分析
当向Flink集群提交用户作业时,从用户角度看,只需要作业处理逻辑正确,输出正确的结果即可;而不用关心作业何时被调度的,作业申请的资源又是如何被分配的以及作业何时会结束;但是了解作业在运行时的具体行为对于我们深入了解Flink原理有非常大的帮助,并且对我们如何编写更合理的作业逻辑有指导意义,因此本文详细分析作业的调度及资源分配以及作业的生命周期。
0 0
【Flink】Flink 底层RPC框架分析
对于Flink中各个组件(JobMaster、TaskManager、Dispatcher等),其底层RPC框架基于Akka实现,本文着重分析Flink中的Rpc框架实现机制及梳理其通信流程。
0 0
从 0 到 1 通过 Flink + Tablestore 进行大数据处理与分析
阿里云实时计算Flink版是一套基于 Apache Flink 构建的⼀站式实时大数据分析平台。在大数据场景下,实时计算 Flink 可提供端到端亚秒级实时数据流批处理能力。表格存储 Tablestore (又名 OTS)是阿里云自研的多模型结构化数据存储,可提供海量结构化数据的存储、查询分析服务。表格存储的双引擎架构支持千万TPS和毫秒级延迟的服务能力,可作为大数据计算的极佳上下游存储。
0 0
从 0 到 1 通过 Flink + Tablestore 进行大数据处理与分析
阿里云实时计算 Flink 版是一套基于 Apache Flink 构建的⼀站式实时大数据分析平台。在大数据场景下,实时计算 Flink 可提供端到端亚秒级实时数据流批处理能力。表格存储 Tablestore (又名 OTS)是阿里云自研的多模型结构化数据存储,可提供海量结构化数据的存储、查询分析服务。表格存储的双引擎架构支持千万TPS和毫秒级延迟的服务能力,可作为大数据计算的极佳上下游存储。
0 0
+关注
文章
问答
文章排行榜
最热
最新
相关电子书
更多
朱翥、贺小令|更快更稳更易用:Flink 自适应批处理能力演
立即下载
任庆盛|Flink CDC + Kafka 加速业务实时化
立即下载
李劲松|Flink Table Store 典型应用场景
立即下载