镇江网站建设镇江,深圳vi设计工作室,赣州微和联网络科技有限公司,如何制作网站的步骤DIM层代码流程图
维度层的重点和难点在于实时电商数仓需要的维度信息一般是动态的变化的#xff0c;并且由于实时数仓一般需要一直运行#xff0c;无法使用常规的配置文件重启加载方式来修改需要读取的ODS层数据#xff0c;因此需要通过Flink-cdc实时监控MySql中的维度数据…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模块 appDimApp里面写的是dim层的具体实现具体步骤如上述流程图所示。function负责存放数据处理的实现类一般会继承相应的父类在dim层可以直接调用这里的子类来实现父类接口让dim层的代码逻辑更加清晰。 realtime-dwd模块如上realtime-dws模块如上 数据清洗ETL
数据清洗简单来说就是对数据进行简单的转换筛选。首先如果在转换过程中出现异常直接过滤掉。注意这里无需抛出异常因为如果throw a exception会导致整个程序异常终止而在数据处理过程中出现部分数据格式错误而无法正常进行格式转换是很常见的只需将异常信息打印到控制台即可。如果转换正常再判断是否满足以下三个条件
数据库名为gmall数据类型不是bootstrap-start或者bootstrap-completedata字段不是null且长度不为0
Flink-cdc读取配置表的数据
Flink中获取数据主要有两个步骤
获取相应的数据源Source 注意在构建Flink-cdc对应的MySQLSource时tableList参数必须是库表.表名结构 调用env.fromSource()方法将数据源的发送过来的数据转换Ds数据流在该方法中可以设置数据的水位线。获取到数据后建议先打印到控制台查看数据的具体结构。注意读取配置信息表的并发度必须设置为1如果不为1只能读取r操作数据其他更新数据无法读取。
public static MySqlSourceString getMySqlSource(String databaseName, String tableName){MySqlSourceString mySqlSource MySqlSource.Stringbuilder().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 代表creater 代表 read需要创建after字段中对应的表u 代表update需要先删除掉旧表然后根据新表的字段创建一个新表 创建HBase连接创建连接是很耗费资源的行为因此新建连接和关闭连接需要写在open和close方法中HBase中想要对表进行创建和删除等DDL操作都由Admin对象管理如果需要对数据进行插入删除等DML操作需要创建Table对象。详细操作细节请看相应代码即可。
public static SingleOutputStreamOperatorTableProcessDim createHbaseTable(DataStreamSourceString mysqlSource) {SingleOutputStreamOperatorTableProcessDim createHBaseTable mysqlSource.flatMap(new RichFlatMapFunctionString, TableProcessDim() {public Connection connection ;Overridepublic void open(Configuration parameters) throws Exception {//获取连接connection HBaseUtil.getHBaseConnection();}Overridepublic void close() throws Exception {//关闭连接HBaseUtil.closeHBaseConn(connection);}Overridepublic void flatMap(String s, CollectorTableProcessDim 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中读取维表配置表信息如果根据当前表的表名查询的状态不为空说明该表为维度数据使用收集器收集起来。
筛选出需要的字段 在维度配置信息表中的sink_column字段里定义了维度表需要的字段使用filter算子对JsonObj里面的data字段进行过滤即可获取到想要的字段数据。
写出到Hbase
过滤后的数据流调用它的addSink方法方法中需要传入一个SinkFunction接口类。该接口需要实现三个方法分别是
open方法获取HBase连接close方法关闭HBase连接invoke方法写入数据时调用的方法根据jsonObj中的type做不同处理如果是delete需要删除对应的维度表数据否则都是直接覆盖写入。 代码的Gitee仓库地址https://gitee.com/langpaian/gmall2023-realtime.git
