跳至主要內容
用Aviator判断风控规则

Aviator自定义函数

在大多数系统中，都是通过Aviator自定义函数来实现关系表达式运算的。

Aviator有两种自定义函数。

  • AbstractFunction：它实现的自定义函数，其call()方法接受1 ~ 20个参数。
添码座原创大约 4 分钟案例大数据风控案例大数据风控FlinkAviator
Flink模式事件匹配

对于那些短时长（统计区间一般在5~15分钟）和浅层操作（注册、登录、修改IP、修改用户名等）的行为，如果使用常规的实时计算方式，例如，从Kafka拉取数据 -> Flink实时计算 -> 指标保存到Redis -> 从Redis读取数据，就显得有些冗长了。

添码座原创大约 2 分钟案例大数据风控案例大数据风控FlinkFlink-CEP
Flink指标采样计算

Redis中不仅保存着Clickhouse计算出来的预聚合数据，也保存Flink计算结果，例如，每5分钟统计一次近1小时的登录数据。

添码座原创大约 17 分钟案例大数据风控案例大数据风控ClickhouseRedisFlink
指标采样设计思路

动态时间片

类似近1小时内用户的登录次数多于3次这样的风控指标，本质上是一种风控关系表达式，它由左变量、关系运算符和右变量（或阈值）组成，如果把它用另外一种方式展现出来就是这样。

风控指标表达式

对于这种风控指标的计算来说，真正的难点不在于计算本身，而在于 如何快速且准确地取得指定时间片的数据。

添码座原创大约 6 分钟案例大数据风控案例大数据风控ClickhouseRedisFlink
Change Data Capture（CDC）

Flink CDC就是一个可以从不同数据源中实时读取数据的技术，它既可以全量读取，也可以增量读取。

它的核心原理是监测并捕获数据库的变动（增删改等），将捕获到的数据发送到数据仓库或者数据湖，也可以写入到消息队列（例如Kafka）供其他服务消费。

引入依赖。

<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-connector-mysql-cdc</artifactId>
    <version>2.3.0</version>
</dependency>
添码座原创大约 1 分钟大数据Flink大数据Flink
Table API & SQL

编程范式

和之前的流式大数据处理一样，在Flink中，所有用于批和流的Table API & SQL也都遵循相同的编程范式，也就是代码上的整体结构都基本上同。

package itechthink.table;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.table.api.*;

/**
 * Table API & SQL编程范式
 *
 */
public class TableApiAndSQLParadigm {
    public static void main(String[] args) throws Exception {
        // 1. 创建批或流执行的Table上下文环境
        Configuration configuration = new Configuration();
        TableEnvironment tableEnv = TableEnvironment.create(configuration);

        // 2.1. 创建数据源表
        tableEnv.createTemporaryTable("SourceTable", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                .column("f0", DataTypes.STRING())
                .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 100L)
                .build());
        // 2.2. 或者，从Table API查询中创建一个Table对象
        Table table1 = tableEnv.from("SourceTable");
        // 2.3. 或者，从SQL查询中创建一个Table对象
        Table table2 = tableEnv.sqlQuery("SELECT * FROM SourceTable");

        // 3.1. 创建一张保存结果的Sink数据表
        tableEnv.executeSql("CREATE TEMPORARY TABLE SinkTable WITH ('connector' = 'blackhole') LIKE SourceTable (EXCLUDING OPTIONS) ");
        // 3.2. 或者，将查询结果保存到Sink数据表
        TableResult tableResult1 = table1.insertInto("SinkTable").execute();
        TableResult tableResult2 = table2.executeInsert("SinkTable");
    }
}
添码座原创大约 10 分钟大数据Flink大数据FlinkTable API
CEP（复杂事件处理）

什么是“CEP”

设想有下面这样的电商应用场景。

  • 找出那些超时未支付的订单，例如，下单10分钟内没有支付的订单有多少？

  • 找出那些1小时内至少有过3次有效交易的用户账户。

  • 找出那些在5秒钟内连续登录失败超过至少3次的账号。

如果说上面的还算简单的话，那下面的这几个可就难办了。

  • 找出行为事件小于8且浏览商品量少于3条的支付订单。

  • 找出识别指定规则的时间并按指定方式输出。

添码座原创大约 9 分钟大数据Flink大数据FlinkCEP
双流JOIN问题

大数据中的关联

在对传统的关系型数据库进行数据查询时，经常会遇到需要将两张或多张表进行关联的操作。

> SELECT a.id, a.name, b.deptid, ... FROM table1 AS a, table2 AS b
  WHERE a.id = b.id;
# 或者
> SELECT a.id, a.name, b.deptid, ... FROM table1 join table2
  ON table1.id = table2.id;
添码座原创大约 5 分钟大数据Flink大数据FlinkJOIN