PyFlink实践与应用

PyFlink场景

Flink 核心能力
- 反欺诈-刷单
- 规则报警
- 运维监控
- 聚合计算-大屏展示
- ETL-日志分析，数据处理
- 个性推荐
- 垃圾分类
Python 生态功能
- 科学计算
- 统计分析
  Flink复杂事件处理（CEP）
  什么是复杂事件处理
  复杂事件处理(CEP，Complex Event Processing)是一种基于动态环境中事件流的分析技术，事件在这里通常是有意义的状态变化，通过分析事件间的关系，利用过滤、关联、聚合等技术，根据事件间的时序关系和聚合关系制定检测规则，持续地从事件流中查询出符合要求的事件序列，最终分析得到更复杂的复合事件，主要用于网络诈欺识别等防止犯罪，银行等金融行业防止，以及风险规避和营销决策等。
  CEP的特征
目标:从有序的简单事件流中发现一些高阶特征;
输入:一个或多个简单事件构成的事件流;
处理:识别简单事件之间的内在联系，多个符合一定规则的简单事件构成复杂事件;
输出:满足规则的复杂事件。

CEP作用

CEP用于分析低延迟、频繁产生的不同来源的事件流。CEP可以帮助在复杂的、不相关的时间流中找出有意义的模式和复杂的关系，以接近实时或准实时的获得通知或组织一些行为。
CEP支持在流上进行模式匹配，根据模式的条件不同，分为连续的条件或不连续的条件;模式的条件允许有时间的限制，当条件范围内没有达到满足的条件时，会导致模式匹配超时。
看起来很简单，但是它有很多不同的功能:

1 输入的流数据，尽快产生结果;
2 在2个事件流上，基于时间进行聚合类的计算;
3 提供实时/准实时的警告和通知;
4 在多样的数据源中产生关联分析模式;
5 高吞吐、低延迟的处理

CEP与SQL语法

事件与事件关联关系

事件应该包含一些基本的要素:类型、发生事件以及更多的一些定义属性
通常需要关联多个事件进行分析处理，其中事件间的关系主要有5种:

时间关联
2. 空间关联
3. 依赖关系
事物的状态属性之间彼此的依赖关系和约束关系。
4. 因果关系

CEP核心组件

Flink 为 CEP 提供了专门的 Flink CEP library，包含如下组件:
• Event Stream
• Pattern定义
• Pattern检测
• 生成Alert

Pattern API
处理事件的规则，被叫作模式(Pattern)。
Flink CEP 提供了 Pattern API 用于对输入流数据进行复杂事件规则定义，用来提取符合规则的事件序列。

Pattern分类
模式大致分为三类:
• 个体模式(Individual Patterns)
• 组成复杂规则的每一个单独的模式定义，就是个体模式
• start.times(3).where(_.behavior.startsWith(‘fav’))
• 组合模式(Combining Patterns，也叫模式序列)
• 很多个体模式组合起来，就形成了整个的模式序列
• 模式序列必须以一个初始模式开始:
• val start = Pattern.begin(‘start’)
• 模式组(Group of Pattern)
• 将一个模式序列作为条件嵌套在个体模式里，成为一组模式。

个体模式(Individual Patterns)
• 个体模式包括单例模式和循环模式。单例模式只接收一个事件，而循环模式可以接收多个事件。
• 量词:可以在一个个体模式后追加量词，也就是指定循环次数

触发条件
每个模式都需要指定触发条件，作为模式是否接受事件进入的判断依据。 CEP中的个体模式主要通过调用 .where()、.or() 和 .until() 来指定条件，按不同的调用方式，可以分成以下几类:
1 简单条件:通过.where()方法对事件中的字段进行判断筛选，决定是否接收该事件 start.where(event=>event.getName.startsWith(“foo”))
2 组合条件:将简单的条件进行合并;or()方法表示或逻辑相连，where的直接组合就相当于与and。
3 终止条件:如果使用了oneOrMore或者oneOrMore.optional，建议使用.until()作为终止条件，以便清理状态。
4 迭代条件:能够对模式之前所有接收的事件进行处理;
调用.where((value,ctx) => {…})，可以调用ctx.getEventForPattern(“name”)

模式序列（组合模式即Combining Patterns）
严格近邻

所有事件按照严格的顺序出现，中间没有任何不匹配的事件，由.next()指定。例如对于模式“a next b”，事件序列“a,c,b1,b2” 没有匹配。

宽松近邻

允许中间出现不匹配的事件，由.followedBy()指定。例如对于模式“a followedBy b”，事件序列“a,c,b1,b2”匹配为{a,b1}。

非确定性宽松近邻

进一步放宽条件，之前已经匹配过的事件也可以再次使用，由.followedByAny()指定。例如对于模式“a followedByAny b”，事件序列“a,c,b1,b2”匹配为{ab1}，{a,b2}。

除了以上模式序列外，还可以定义“不希望出现某种近邻关系”:

.notNext():不想让某个事件严格紧邻前一个事件发生。
.notFollowedBy():不想让某个事件在两个事件之间发生。

模式检测

指定要查找的模式序列后，就可以将其应用于输入流以检测潜在匹配。
调用 CEP.pattern()，给定输入流和模式，就能得到一个 PatternStream。

匹配事件的提取

创建 PatternStream 之后，就可以应用 select 或者 flatSelect 方法，从检测到的事件序列中提取事件

Alink机器学习框架介绍和使用

机器学习框架

Alink 是基于 Flink 的机器学习算法平台
同时支持批式/流式算法，提供丰富的算法库
帮助数据分析和应用开发人员能够从数据处理，特征工程，模型训练，预测，端到端完成整个流程
2019年11月 Flink Forward Asia 上宣布开源
项目地址:https://github.com/alibaba/Alink
开源算法列表

Alink VS SparkML

开源数据源

构建一个Alink项目
第一步:创建 Maven 项目
第二步:修改 Pom文件，导入 Alink 相关 Jar 包
第三步:构建运行

KMeans Java代码实例

运行Alink项目
快速开始在集群上运行Alink算法
1. 准备Flink集群
$ wget https://archive.apache.org/dist/flink/flink-1.11.0/flink-1.10.0-bin- scala_2.11.tgz
$ tar -xf flink-1.11.0-bin-scala_2.11.tgz && cd flink-1.11.0 $ ./bin/start-cluster.sh
2. 准备Alink算法包
$ git clone https://github.com/alibaba/Alink.git
$ cd Alink && mvn -Dmaven.test.skip=true clean package shade:shade
3. 运行Java示例
$ ./bin/flink run -p 1 -c com.alibaba.alink.ALSExample alink_examples-1.1- SNAPSHOT.jar