数仓开发之DIM层

目录

一：DIM层设计要点

 二：DIM层大概实操流程

     2.1 读取数据

   2.2 过滤数据

   2.3 写出数据

 三：配置表

3.1 配置表设计

 四：实操流程

4.1 接收Kafka数据，过滤空值数据

4.2 动态拆分维度表功能

4.3  把流中的数据保存到对应的维度表

五：具体代码实现

 5.1 接收Kafka数据，过滤空值数据

5.2 根据MySQL的配置表，动态进行分流

5.3 保存维度到HBase(Phoenix)

---

一：DIM层设计要点

（1）DIM层的设计依据是维度建模理论，该层存储维度模型的维度表。

（2）DIM层的数据存储在 HBase 表中。

DIM 层表是用于维度关联的，要通过主键去获取相关维度信息，这种场景下 K-V 类型数据库的效率较高。常见的 K-V 类型数据库有 Redis、HBase，而 Redis 的数据常驻内存，会给内存造成较大压力，因而选用 HBase 存储维度数据。

（3）DIM层表名的命名规范为dim_表名

 二：DIM层大概实操流程

     2.1 读取数据

Kafka---topic_db(包含所有的46张业务表)

   2.2 过滤数据

过滤出所需要的维表数据
        过滤条件:在代码中给定十几张维表的表名
        问题:如果增加维表,需要修改代码-重新编译-打包-上传、重启任务
        优化1:不修改代码、只重启任务
            配置信息中保存需要的维表信息,配置信息只在程序启动的时候加载一次
        优化2:不修改代码、不重启任务
            方向:让程序在启动以后还可以获取配置信息中增加的内容
            具体实施:
                1) 定时任务:每隔一段时间加载一次配置信息
                    将定时任务写在Open方法
                2) 监控配置信息:一旦配置信息增加了数据,可以立马获取到
                    (1) MySQLBinlog:FlinkCDC监控直接创建流
                        a.将配置信息处理成广播流:缺点 -> 如果配置信息过大,冗余太多
                        b.按照表名进行KeyBy处理:缺点 -> 有可能产生数据倾斜
                    (2) 文件:Flume->Kafka->Flink消费创建流

   2.3 写出数据

将数据写出到Phoenix、JdbcSink、自定义Sink

 三：配置表

本层的任务是将业务数据直接写入到不同的 HBase 表中。那么如何让程序知道流中的哪些数据是维度数据？维度数据又应该写到 HBase 的哪些表中？为了解决这个问题，我们选择在 MySQL 中构建一张配置表，通过 Flink CDC 将配置表信息读取到程序中。

3.1 配置表设计

1）字段解析

我们将为配置表设计五个字段

• source_table：作为数据源的业务数据表名 
• sink_table：作为数据目的地的 Phoenix 表名
• sink_columns：Phoenix 表字段
• sink_pk：Phoenix 表主键
• sink_extend：Phoenix 建表扩展，即建表时一些额外的配置语句

将 source_table 作为配置表的主键，可以通过它获取唯一的目标表名、字段、主键和建表扩展，从而得到完整的 Phoenix 建表语句。

 数据格式：

{"before":null,"after":
{"source_table":"aa","sink_table":"bb","sink_columns":"cc","sink_pk":"id","sink_extend":"xxx"},"source":
{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":165251303
9549,"snapshot":"false","db":"gmall-211126-
config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0
,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1652513039551,"transaction":null}

2）在Mysql中创建数据库建表并开启Binlog

（1）创建数据库 gmall_config ，注意:和 gmall 业务库区分开

[atguigu@hadoop102 db_log]$ mysql -uroot -p000000 -e"create database gmall_config charset 
utf8 default collate utf8_general_ci"

（2）在 gmall_config 库中创建配置表 table_process

CREATE TABLE `table_process` (
  `source_table` varchar(200) NOT NULL COMMENT '来源表',
  `sink_table` varchar(200) DEFAULT NULL COMMENT '输出表',
  `sink_columns` varchar(2000) DEFAULT NULL COMMENT '输出字段',
  `sink_pk` varchar(200) DEFAULT NULL COMMENT '主键字段',
  `sink_extend` varchar(200) DEFAULT NULL COMMENT '建表扩展',
  PRIMARY KEY (`source_table`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

（3）在MySQL配置文件中增加 gmall_config 开启Binlog

[axing@hadoop107 ~]$ sudo vim /etc/my.cnf

 （4）为了方便测试，目前就插入两张表名数据，作为维度表

 四：实操流程

4.1 接收Kafka数据，过滤空值数据

对Maxwell抓取的数据进行ETL，有用的部分保留，没用的过滤掉。

4.2 动态拆分维度表功能

由于Maxwell是把全部数据统一写入一个Topic中, 这样显然不利于日后的数据处理。所以需要把各个维度表拆开处理。

在实时计算中一般把维度数据写入存储容器，一般是方便通过主键查询的数据库比如HBase,Redis,MySQL等。

这样的配置不适合写在配置文件中，因为这样的话，业务端随着需求变化每增加一张维度表表，就要修改配置重启计算程序。

所以这里需要一种动态配置方案，把这种配置长期保存起来，一旦配置有变化，实时计算可以自动感知。这种可以有三个方案实现：

一种是用Zookeeper存储，通过Watch感知数据变化；

另一种是用mysql数据库存储，周期性的同步；

再一种是用mysql数据库存储，使用广播流。

这里选择第三种方案，主要是MySQL对于配置数据初始化和维护管理，使用FlinkCDC读取配置信息表，将配置流作为广播流与主流进行连接。

4.3  把流中的数据保存到对应的维度表

 维度数据保存到HBase的表中。

五：具体代码实现

 5.1 接收Kafka数据，过滤空值数据

1）创建 KafkaUtil 工具类

和 Kafka 交互要用到 Flink 提供的 FlinkKafkaConsumer、FlinkKafkaProducer 类，为了提高模板代码的复用性，将其封装到 KafkaUtil 工具类中。

此处从 Kafka 读取数据，创建 getKafkaConsumer(String topic, String groupId) 方法

public class KafkaUtil {
    static String BOOTSTRAP_SERVERS = "hadoop102:9092, hadoop103:9092, hadoop104:9092";
    static String DEFAULT_TOPIC = "default_topic";
 
    public static FlinkKafkaConsumer  getKafkaConsumer(String topic, String groupId) {
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
 
        FlinkKafkaConsumer consumer = new FlinkKafkaConsumer<>(topic,
                new KafkaDeserializationSchema() {
                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }
 
                    @Override
                    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        if(record != null && record.value() != null) {
                            return new String(record.value());
                        }
                        return null;
                    }
 
                    @Override
                    public TypeInformation getProducedType() {
                        return TypeInformation.of(String.class);
                    }
                }, prop);
        return consumer;
}
}

2）主程序

package com.atguigu.app.dim;
 
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.TableProcessFunction;
import com.atguigu.bean.TableProcess;
import com.atguigu.utils.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
 
public class DimApp {
    public static void main(String[] args) throws Exception {
 
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);//生产环境，并行度应设置为kafka主题的分区数
 
 
        /*
        //生产环境下使用：
        //1.1 开启checkpoint
        env.enableCheckpointing(5*6000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10*6000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
        //1.2 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop107:8020/211126/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");
         */
 
        //2.读取kafka topic_db主题数据创建主流
        String topic ="topic_db";
        String groupId = "dim_app_211126";
        DataStreamSource kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
 
        //3.过滤掉非JSON数据以及保留新增、变化以及初始化数据并将数据转换为JSON格式
        SingleOutputStreamOperator filterJsonDS = kafkaDS.flatMap(new FlatMapFunction() {
            @Override
            public void flatMap(String value, Collector collector) throws Exception {
 
                try {
                    //将数据装换为JSON格式
                    JSONObject jsonObject = JSON.parseObject(value);
                    //获取数据中的操作类型字段
                    String type = jsonObject.getString("type");
                    //保留新增、变化、以及初始化数据
                    if ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type)) {
                        collector.collect(jsonObject);
                    }
                } catch (Exception e) {
                    System.out.println("脏数据：" + value);//或者写入侧输出流
                }
 
            }
        });
 
        //4.使用FlinkCDC读取mysql配置信息表创建配置流
 
        MySqlSource mySqlSource = MySqlSource.builder()
                .hostname("hadoop107")
                .port(3306)
                .username("root")
                .password("000000")
                .databaseList("gmall-config")
                .tableList("gmall-config.table_process")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();
 
        DataStreamSource mysqlSourceDS = env.fromSource(mySqlSource,
                WatermarkStrategy.noWatermarks(),
                "MysqlSource"
        );
 
        //5.将配置流处理为广播流
        MapStateDescriptor mapStateDescriptor = new MapStateDescriptor<>("map-state", String.class, TableProcess.class);
        BroadcastStream broadcastStream = mysqlSourceDS.broadcast(mapStateDescriptor);
 
        //6.连接主流和广播流
        BroadcastConnectedStream connectedStream = filterJsonDS.connect(broadcastStream);
 
        //7.处理连接流，根据配置信息处理主流数据
        SingleOutputStreamOperator dimDS = connectedStream.process(new TableProcessFunction(mapStateDescriptor));
 
        //8.将数据写出到Phoenix
        dimDS.print(">>>>>");
 
        //9.启动任务
        env.execute();
    }
}

5.2 根据MySQL的配置表，动态进行分流

1）导入依赖

<dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-connector-jdbc_${scala.version}artifactId>
            <version>${flink.version}version>
        dependency>
 
        <dependency>
            <groupId>com.ververicagroupId>
            <artifactId>flink-connector-mysql-cdcartifactId>
            <version>2.1.0version>
        dependency>
 
        <dependency>
            <groupId>org.apache.phoenixgroupId>
            <artifactId>phoenix-sparkartifactId>
            <version>5.0.0-HBase-2.0version>
            <exclusions>
                <exclusion>
                    <groupId>org.glassfishgroupId>
                    <artifactId>javax.elartifactId>
                exclusion>
            exclusions>
        dependency>
 
	
		<dependency>
            <groupId>org.apache.flinkgroupId>
            <artifactId>flink-table-api-java-bridge_2.12artifactId>
            <version>1.13.0version>
        dependency>

2）创建配置表实体类

package com.atguigu.gmall.realtime.bean;
import lombok.Data;
 
@Data
public class TableProcess {
    //来源表
    String sourceTable;
    //输出表
    String sinkTable;
    //输出字段
    String sinkColumns;
    //主键字段
    String sinkPk;
    //建表扩展
    String sinkExtend;
}

3）编写操作读取配置表形成广播流

// TODO 6. FlinkCDC 读取配置流并广播流
        // 6.1 FlinkCDC 读取配置表信息
        MySqlSource mySqlSource = MySqlSource.builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall_config") // set captured database
                .tableList("gmall_config.table_process") // set captured table
                .username("root")
                .password("000000")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.initial())
                .build();
 
        // 6.2 封装为流
        DataStreamSource mysqlDSSource = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MysqlSource");
 
        // 6.3 广播配置流
        MapStateDescriptor tableConfigDescriptor = new MapStateDescriptor("table-process-state", String.class, TableProcess.class);
        BroadcastStream broadcastDS = mysqlDSSource.broadcast(tableConfigDescriptor);
 
        // TODO 7. 连接流
        BroadcastConnectedStream connectedStream = filterDS.connect(broadcastDS);

4）定义一个项目中常用的配置常量类GmallConfig

package com.atguigu.gmall.realtime.common;
 
public class GmallConfig {
 
    // Phoenix库名
    public static final String HBASE_SCHEMA = "GMALL2022_REALTIME";
 
    // Phoenix驱动
    public static final String PHOENIX_DRIVER = "org.apache.phoenix.jdbc.PhoenixDriver";
 
    // Phoenix连接参数
    public static final String PHOENIX_SERVER = "jdbc:phoenix:hadoop102,hadoop103,hadoop104:2181";
}

5）自定义函数MyBroadcastFunction

（1）定义类MyBroadcastFunction

package com.atguigu.gmall.realtime.app.func;
 
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
 
public class MyBroadcastFunction extends BroadcastProcessFunction {
 
	private MapStateDescriptor tableConfigDescriptor;
 
public MyBroadcastFunction(MapStateDescriptor tableConfigDescriptor) {
        this.tableConfigDescriptor = tableConfigDescriptor;
}
 
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext readOnlyContext, Collector out) throws Exception {
 
    }
 
    @Override
    public void processBroadcastElement(String jsonStr, Context context, Collector out) throws Exception {
 
    }
}

 （2）自定义函数MyBroadcastFunction-open

// 定义Phoenix的连接
private Connection conn;
 
@Override
    public void open(Configuration parameter) throws Exception {
        super.open(parameter);
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

（3）自定义函数MyBroadcastFunction-processBroadcastElement

@Override
    public void processBroadcastElement(String jsonStr, Context context, Collector out) throws Exception {
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        BroadcastState tableConfigState = context.getBroadcastState(tableConfigDescriptor);
        String op = jsonObj.getString("op");
        if ("d".equals(op)) {
            TableProcess before = jsonObj.getObject("before", TableProcess.class);
            String sourceTable = before.getSourceTable();
            tableConfigState.remove(sourceTable);
        } else {
            TableProcess config = jsonObj.getObject("after", TableProcess.class);
 
            String sourceTable = config.getSourceTable();
            String sinkTable = config.getSinkTable();
            String sinkColumns = config.getSinkColumns();
            String sinkPk = config.getSinkPk();
            String sinkExtend = config.getSinkExtend();
 
            tableConfigState.put(sourceTable, config);
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
        }
    }

（4）自定义函数MyBroadcastFunction-checkTable

在 Phoenix 建表之前要先创建命名空间 GMALL2022_REALTIM。

0: jdbc:phoenix:> create schema GMALL2022_REALTIME;

checkTable() 方法如下

/**
     * Phoenix 建表函数
     *
     * @param sinkTable 目标表名  eg. test
     * @param sinkColumns 目标表字段  eg. id,name,sex
     * @param sinkPk 目标表主键  eg. id
     * @param sinkExtend 目标表建表扩展字段  eg. ""
     *                   eg. create table if not exists mydb.test(id varchar primary key, name varchar, sex varchar)...
     */
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        // 封装建表 SQL 
        StringBuilder sql = new StringBuilder();
        sql.append("create table if not exists " + GmallConfig.HBASE_SCHEMA
                + "." + sinkTable + "(\n");
        String[] columnArr = sinkColumns.split(",");
        // 为主键及扩展字段赋默认值
        if (sinkPk == null) {
            sinkPk = "id";
        }
        if (sinkExtend == null) {
            sinkExtend = "";
        }
        // 遍历添加字段信息
        for (int i = 0; i < columnArr.length; i++) {
            sql.append(columnArr[i] + " varchar");
            // 判断当前字段是否为主键
            if (sinkPk.equals(columnArr[i])) {
                sql.append(" primary key");
            }
            // 如果当前字段不是最后一个字段，则追加","
            if (i < columnArr.length - 1) {
                sql.append(",\n");
            }
        }
        sql.append(")");
        sql.append(sinkExtend);
        String createStatement = sql.toString();
        // 为数据库操作对象赋默认值，执行建表 SQL 
        PreparedStatement preparedSt = null;
        try {
            preparedSt = conn.prepareStatement(createStatement);
            preparedSt.execute();
        } catch (SQLException sqlException) {
            sqlException.printStackTrace();
            System.out.println("建表语句\n" + createStatement + "\n执行异常");
        } finally {
            if (preparedSt != null) {
                try {
                    preparedSt.close();
                } catch (SQLException sqlException) {
                    sqlException.printStackTrace();
                    throw new RuntimeException("数据库操作对象释放异常");
                }
            }
        }
    }

（5）自定义函数MyBroadcastFunction-processElement()

@Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext readOnlyContext, Collector out) throws Exception {
        ReadOnlyBroadcastState tableConfigState = readOnlyContext.getBroadcastState(tableConfigDescriptor);
        // 获取配置信息
        String sourceTable = jsonObj.getString("table");
        TableProcess tableConfig = tableConfigState.get(sourceTable);
        if (tableConfig != null) {
            JSONObject data = jsonObj.getJSONObject("data");
            String sinkTable = tableConfig.getSinkTable();
 
            // 根据 sinkColumns 过滤数据
            String sinkColumns = tableConfig.getSinkColumns();
            filterColumns(data, sinkColumns);
 
            // 将目标表名加入到主流数据中
            data.put("sinkTable", sinkTable);
			
            out.collect(data);
        }
    }

（6）自定义函数MyBroadcastFunction-filterColumns()，校验字段，过滤掉多余的字段

private void filterColumns(JSONObject data, String sinkColumns) {
        Set> dataEntries = data.entrySet();
        dataEntries.removeIf(r -> !sinkColumns.contains(r.getKey()));
    }

（7）主程序DimSinkApp中调用MyBroadcastFunction提取维度数据

// TODO 8. 处理维度表数据
        SingleOutputStreamOperator dimDS = connectedStream.process(
                new MyBroadcastFunction(tableConfigDescriptor)
        );

5.3 保存维度到HBase(Phoenix)

1）程序流程分析

 

DimSink 继承了RickSinkFunction，这个function得分两条时间线：

一条是任务启动时执行open操作（图中紫线），我们可以把连接的初始化工作放在此处一次性执行；

另一条是随着每条数据的到达反复执行invoke()（图中黑线）,在这里面我们要实现数据的保存，主要策略就是根据数据组合成sql提交给hbase。

2）创建 PhoenixUtil 工具类，在其中创建insertValues()方法

package com.atguigu.gmall.realtime.util;
 
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.StringUtils;
 
import java.sql.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
 
public class PhoenixUtil {
    /**
     * Phoenix 表数据导入方法
     *
     * @param conn 连接对象
     * @param sinkTable 写入数据的 Phoenix 目标表名
     * @param data      待写入的数据
     */
    public static void insertValues(Connection conn, String sinkTable, JSONObject data) {
        // 获取字段名
        Set columns = data.keySet();
        // 获取字段对应的值
        Collection