电商风控系统（flink+groovy+flume+kafka+redis+clickhouse+mysql）

一.项目概览

电商的防止薅羊毛的风控系统

需要使用 groovy 进行风控规则引擎的编写然后其它技术进行各种数据的存储及处理

薅羊毛大致流程

如果单纯使用 if else在业务代码中进行风控规则的编写那么维护起来会比较麻烦并且跟业务系统强绑定不合适所以一般独立成一个单独的系统

常见风控规则列举

风控引擎设计的核心点

业务逻辑概览

事件接入中心

技术架构

分层

各单位占比

二.flink常见知识点实战

从下图可以看出跟之前yarn类似还是有管理有大领导校领导打工人打工人来执行任务

分别对应 jobmamager taskmanager taskslot 由 taskslot 执行任务每个

2.1state

实战

首先看个入门级代码就是对 字符串的出现次数的结果进行实时统计与打印

package com.juege.hope.opentech.flinktest;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class FlinkTurotial1_17 {

    public static void main(String[] args) throws Exception {

        //todo 1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //todo 2.读取数据
        DataStreamSource<String> stringDataStreamSource = env.readTextFile("D:\\juege\\code\\hope-backend\\opentech\\src\\main\\resources\\flinkTextSource.txt");

        //todo 3.进行数据处理 先 flatmap 再 keyby 再 sum 再打印输出
        stringDataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = s.split(" ");
                for (String word : words) {
                    collector.collect(new Tuple2<>(word, 1));
                }
            }
        }).keyBy(0).sum(1).print();

        //todo 4.执行任务
        env.execute("pantouyu");
    }

}

数据源

显示结果如下

使用state来实现sum方法的效果

package com.example.flinktest.test;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class FlinkTurotial1_17 {

    public static void main(String[] args) throws Exception {

        //todo 1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //todo 2.读取数据
        DataStreamSource<String> stringDataStreamSource = env.readTextFile("D:\\juege\\code\\flink-test\\src\\main\\resources\\flinkTextSource.txt");

        //todo 3.进行数据处理 先 flatmap 再 keyby 再 sum 再打印输出
        stringDataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = s.split(" ");
                for (String word : words) {
                    collector.collect(new Tuple2<>(word, 1));
                }
            }
        }).keyBy(0)
                .flatMap(new SumFunction()).print();
//                .sum(1).print();

        //todo 4.执行任务
        env.execute("pantouyu");
    }

}

package com.example.flinktest.test;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class SumFunction extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {
    private transient ValueState<Integer> sumState;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("sumState", Integer.class);
        sumState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, Integer>> out) throws Exception {
        String key = value.f0;
        Integer inputValue = value.f1;

        Integer currentSum = sumState.value();
        if (currentSum == null) {
            currentSum = 0;
        }

        currentSum += inputValue;
        sumState.update(currentSum);

        out.collect(new Tuple2<>(key, currentSum));
    }
}