当前位置: 首页 > news >正文

网站做图分辨率是多少合适那种转转假网站怎么做的

网站做图分辨率是多少合适,那种转转假网站怎么做的,服装 营销型网站案例,开封网站建设优化DIM层代码流程图 维度层的重点和难点在于实时电商数仓需要的维度信息一般是动态的变化的,并且由于实时数仓一般需要一直运行,无法使用常规的配置文件重启加载方式来修改需要读取的ODS层数据,因此需要通过Flink-cdc实时监控MySql中的维度数据…

DIM层代码流程图

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

  1. 消费Kafka ods业务主题数据
  2. 数据清洗:是否为JSON格式
  3. 使用flink-cdc读取监控配置表数据
  4. 在HBase中创建维度表
  5. 做成广播流
  6. 连接主流和广播流
  7. 筛选出需要写出的字段
  8. 写出到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会导致整个程序异常终止,而在数据处理过程中出现部分数据格式错误而无法正常进行格式转换是很常见的,只需将异常信息打印到控制台即可。如果转换正常,再判断是否满足以下三个条件:

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

Flink-cdc读取配置表的数据

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

  1. 获取相应的数据源Source
    • 注意:在构建Flink-cdc对应的MySQLSource时,tableList参数必须是库表.表名结构
  2. 调用env.fromSource()方法将数据源的发送过来的数据转换Ds数据流,在该方法中可以设置数据的水位线。
  3. 获取到数据后,建议先打印到控制台查看数据的具体结构。
  4. 注意读取配置信息表的并发度必须设置为1;如果不为1,只能读取r操作数据,其他更新数据无法读取。
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 ;@Overridepublic void open(Configuration parameters) throws Exception {//获取连接connection = HBaseUtil.getHBaseConnection();}@Overridepublic void close() throws Exception {//关闭连接HBaseUtil.closeHBaseConn(connection);}@Overridepublic 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表决定了后续程序筛选哪个表作为维度信息,并且定义了表中有哪些字段。

  1. 转换为广播流只需要调用上述得到的TableProcessDimStream的broadcast方法
  2. 使用的主流(gmall业务数据)的connect方法,得到一个连接流,然后对连接流进行process处理。
  3. 创建BroadcastProcessFunction,在里面分别有两个函数
    • processBroadcastElement():处理广播流数据
    • processElement():处理主流数据
  4. 广播流处理逻辑:
    • 读取广播状态
    • 将配置表信息写到广播状态中
    • 根据广播状态数据的op对状态做相应的修改
  5. 主流处理逻辑:
    • 查询广播状态,判断当前数据对应的表是否存在于状态中
    • 如果数据比状态来的更早,造成状态为空,需要对状态做预处理(提前从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

http://www.yayakq.cn/news/604681/

相关文章:

  • 查看网站开发网站域名备案服务
  • 网站美工设计公司电话怎么做能让网站尽快收录
  • 如何设计网站建设方案网站建设方案书 icp备案
  • 那个网站可以做软件出售的网站建设的功能需求分析
  • 服务器上 网站上海php网站建设
  • 攀枝花网站建设wordpress 模板 怎么用
  • 餐饮门户网站 方案怎么做百度投诉电话人工服务总部
  • 市建设局网站施工企业跨专业接工作是否有效
  • wordpress 直播网站吗网站开发教学视频
  • php做网站示例网站开发工作前景
  • 汕头网站优化电话青浦建设网站公司
  • 互动网站设计wordpress安装在本地安装
  • 求个网站2020急急急胶州建网站
  • 云南建网站做消费金融网站价格
  • 滁州网站建设工作室h5链接是什么意思
  • 湛江网站设计哪家好二手域名
  • 快速网站如何做国外的电商网站
  • 百度头条怎么做网站公众号开发教程零基础
  • 手机怎样做自己的网站无锡企业网站的建设
  • 湖北鼎天宏图建设工程有限公司网站东道品牌创意集团
  • 如何分析对手网站关键词网站utf8乱码
  • 湖北建设银行网站首页行业门户网站是什么
  • .net给网站做短信验证网站首页logo怎么修改
  • 网站引导页一般是什么格式网站如何接入支付宝
  • 企业网站报备网站备案 拍照网点
  • app网站建设介绍上海网站seo设计
  • 北京企业网站优化过年做那些网站致富
  • joomla适合做什么网站公司规模介绍范文
  • 免费推广网站地址大全网站如何提升用户体验
  • 网站开发脚本语言opencms wordpress