Flink电商实时数仓（三）

DIM层代码流程图

维度层的重点和难点在于实时电商数仓需要的维度信息一般是动态的变化的，并且由于实时数仓一般需要一直运行，无法使用常规的配置文件重启加载方式来修改需要读取的ODS层数据，因此需要通过Flink-cdc实时监控MySql中的维度数据配置信息表，实时动态的发布广播信息。主流数据根据广播数据及时调整处理逻辑，并自动在HBase中创建相应的维度表和写入相应的维度数据。

消费Kafka ods业务主题数据
数据清洗：是否为JSON格式
使用flink-cdc读取监控配置表数据
在HBase中创建维度表
做成广播流
连接主流和广播流
筛选出需要写出的字段
写出到Hbase

在这里插入图片描述

整体架构

realtime-common模块
- base: 所有Flink程序的基类，负责搭建Flink运行环境和设置并行度和检查点等相关参数。其中我们的数据来源也确定为Kafka，故数据源代码也写在这里。每个Flink程序的具体处理逻辑由handle()函数来负责处理。
- bean：负责存放项目运行过程中需要用到的bean对象，比如当前flink-cdc程序中需要用到的TableProcessDim类，配置信息表对象。
- constant：负责存放程序中需要使用到常量参数
- function：负责存放一些通用的函数方法
- util：一般存放和数据连接相关的工具类
- test目录: 用来在写正式代码前测试连接是否通畅，数据是否可以正常发送。
realtime-dim模块
- app：DimApp里面写的是dim层的具体实现，具体步骤如上述流程图所示。
- function：负责存放数据处理的实现类，一般会继承相应的父类，在dim层可以直接调用这里的子类来实现父类接口，让dim层的代码逻辑更加清晰。
realtime-dwd模块：如上
realtime-dws模块：如上

在这里插入图片描述

数据清洗ETL

数据清洗，简单来说就是对数据进行简单的转换筛选。首先如果在转换过程中出现异常，直接过滤掉。注意这里无需抛出异常，因为如果throw a exception会导致整个程序异常终止，而在数据处理过程中出现部分数据格式错误而无法正常进行格式转换是很常见的，只需将异常信息打印到控制台即可。如果转换正常，再判断是否满足以下三个条件：

数据库名为gmall
数据类型不是bootstrap-start或者bootstrap-complete
data字段不是null且长度不为0

Flink-cdc读取配置表的数据

Flink中获取数据主要有两个步骤：

获取相应的数据源Source
- 注意：在构建Flink-cdc对应的MySQLSource时，tableList参数必须是库表.表名结构
调用env.fromSource()方法将数据源的发送过来的数据转换Ds数据流，在该方法中可以设置数据的水位线。
获取到数据后，建议先打印到控制台查看数据的具体结构。

public static MySqlSource<String> getMySqlSource(String databaseName, String tableName){
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_PORT)
                .username(Constant.MYSQL_USER_NAME)
                .password(Constant.MYSQL_PASSWORD)
                .databaseList(databaseName) // set captured database
                .tableList(databaseName+"."+tableName) // set captured table

                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.initial())
                .build();

        return mySqlSource;
    }

在HBase中创建维度表

数据库中的配置表数据经过Flink-cdc处理后发送到这里是json格式的字符串，这里根据数据的四种类型op在HBase中进行不同的建表删表操作，同时对数json字符数据进行转换映射处理，转换为对应的bean对象数据流。这里一个数据产生一个处理后的对象，故使用Map算子或FlatMap算子都可以。

op类型
- d 代表delete，需要删除before字段中对应的表
- c 代表create，r 代表 read，需要创建after字段中对应的表
- u 代表update，需要先删除掉旧表，然后根据新表的字段创建一个新表
创建HBase连接，创建连接是很耗费资源的行为，因此新建连接和关闭连接需要写在open和close方法中
HBase中想要对表进行创建和删除等DDL操作，都由Admin对象管理；如果需要对数据进行插入删除等DML操作，需要创建Table对象。详细操作细节请看相应代码即可。

public static SingleOutputStreamOperator<TableProcessDim> createHbaseTable(DataStreamSource<String> mysqlSource) {
        SingleOutputStreamOperator<TableProcessDim> createHBaseTable = mysqlSource.flatMap(
                new RichFlatMapFunction<String, TableProcessDim>() {


                    public Connection connection ;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //获取连接
                        connection = HBaseUtil.getHBaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        //关闭连接
                        HBaseUtil.closeHBaseConn(connection);
                    }

                    @Override
                    public void flatMap(String s, Collector<TableProcessDim> out){
                        //使用读取的配置表数据，到HBase中创建与之对应的表格

                        try {
                            JSONObject jsonObject = JSONObject.parseObject(s);
                            String op = jsonObject.getString("op");
                            TableProcessDim dim;//维度表
                            if ("d".equals(op)) {
                                dim = jsonObject.getObject("before", TableProcessDim.class);
                                dim.setOp(op);
                                //当配置表发送一个D类型的数据，对应的HBase需要删除一张维度表
                                deleteTable(dim);
                            } else if ("c".equals(op) || "r".equals(op)) {
                                dim = jsonObject.getObject("after", TableProcessDim.class);
                                createTable(dim);
                                dim.setOp(op);
                            } else {//op = 'u', 即修改
                                dim = jsonObject.getObject("after", TableProcessDim.class);
                                deleteTable(dim);
                                createTable(dim);
                            }
                            dim.setOp(op);
                            out.collect(dim);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }

                    private void createTable(TableProcessDim dim) {
                        String sinkFamily = dim.getSinkFamily();
                        String[] split = sinkFamily.split(",");
                        try {
                            HBaseUtil.createHBaseTable(connection,Constant.HBASE_NAMESPACE,dim.getSinkTable(),split);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }

                    }

                    private void deleteTable(TableProcessDim dim) {
                        try {
                            HBaseUtil.dropHBaseTable(connection, Constant.HBASE_NAMESPACE, dim.getSinkTable());
                        } catch (IOException e) {
                            e.printStackTrace();
                        }

                    }
                }
        );
        return createHBaseTable;
    }

主流连接广播流

从Flink-cdc获取的数据（gmall2023_config）是作为一个参数来控制我们对于主流即ODS层数据（gmall数据库的业务数据）的处理逻辑。gmall2023)_config库中的Table_process_dim表决定了后续程序筛选哪个表作为维度信息，并且定义了表中有哪些字段。

转换为广播流只需要调用上述得到的TableProcessDimStream的broadcast方法
使用的主流(gmall业务数据)的connect方法，得到一个连接流，然后对连接流进行process处理。
创建BroadcastProcessFunction，在里面分别有两个函数
- processBroadcastElement():处理广播流数据
- processElement()：处理主流数据
广播流处理逻辑：
- 读取广播状态
- 将配置表信息写到广播状态中
- 根据广播状态数据的op对状态做相应的修改
主流处理逻辑：
- 查询广播状态，判断当前数据对应的表是否存在于状态中
- 如果数据比状态来的更早，造成状态为空，需要对状态做预处理（提前从mysql中读取维表配置表信息）
- 如果根据当前表的表名查询的状态不为空，说明该表为维度数据，使用收集器收集起来。