3、构建实时数据仓库-ods和dim层构建

3、构建实时数据仓库

项目平台搭建架构及其总体流程

1、flink整合hive的catalog

因为本项目中的对应kafka中的表都存在hive的元数据库中，所以需要创建一个hive的catalog,方便直接操作

-- 进入sql客户端
sql-client.sh 
 
-- 创建hive catalog
CREATE CATALOG hive_catalog WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/usr/local/soft/hive-1.2.1/conf'
);
-- set the HiveCatalog as the current catalog of the session
USE CATALOG hive_catalog;
 
 
-- 创建一个flink_init.sql文件，将hive catalog放进去，后面再启动sql-client时指定sql文件

2、在hviecatalog中创建数据库

离线数据仓库：每一个库对应一个用户，每一个库对应hdfs中一个目录

实时数据仓库每一个层的数据保存在不同位置，ods,dwd,dws保存在kafka，dim,ads的数据保存在数据库中

create database gma_ods;
create database gma_dwd;
create database gma_dws;
create database gma_dim;
create database gma_ads;

4、数据采集

必须现在mysql中创建数据库，然后打开canal,实时监控数据库，当我们对于这个库做任何操作，kafka中就会实时同步我们的操作(通过json格式)

# 1、在mysql中创建数据库gma，指定编码格式为utf-8
# 2、修改canal配置文件
cd /usr/local/soft/canal/conf/example
# 修改配置文件
vim instance.properties
# 增加动态topic配置，每个表在kafka中生成一个topic
canal.mq.dynamicTopic=gma\\..*
 
 
# 3、开启mysqlbinlog日志
vim /etc/my.cnf 
# 在配置文件中增加二配置
# 需要将配置放在[mysqld]后面
# 打开binlog
log-bin=mysql-bin
# 选择ROW(行)模式
binlog-format=ROW
# 重启mysql
service mysqld restart
 
# 4、启动canal
cd /usr/local/soft/canal/bin
./restart.sh
 
# 5、在数据库中创建表
# 执行下面这个sql文件
init_mysql_table.sql
 
# 6、查看topic是否生成
kafka-topics.sh --list  --zookeeper  master:2181
# 如果topic没有生成，检测前面哪里出了问题，创建表之后必须要有topic
 
# 7、导入数据到mysql中
# 执行sql文件
load_data.sql
 
# 8、使用kafka控制台消费者消费数据
kafka-console-consumer.sh --bootstrap-server  master:9092 --from-beginning --topic gma.base_category1

5、ODS层

• 在flink sql中创建ods的表

# 进入sql-client
sql-client.sh -i flink_init.sql
 
# 创建ods层所有的表
ods_mysql_kafka_base_category1：商品一级分类表
ods_mysql_kafka_base_category2：商品二级分类表
ods_mysql_kafka_base_category3：商品三级分类表
ods_mysql_kafka_base_province：省份配置报表
ods_mysql_kafka_base_region：地区配置表
ods_mysql_kafka_base_trademark：品牌表
ods_mysql_kafka_date_info：时间配置表
ods_mysql_kafka_holiday_info：节假日表
ods_mysql_kafka_holiday_year：节假日对应时间表
ods_mysql_kafka_order_detail：订单详情表，一个订单中一个商品一条数据
ods_mysql_kafka_order_info：订单表。一个订单一条数据，订单表中有订单状态
ods_mysql_kafka_order_status_log：订单转台变化日志记录表
ods_mysql_kafka_payment_info：支付流水表
ods_mysql_kafka_sku_info：商品信息表
ods_mysql_kafka_user_info：用户信息表

6、DIM层

每个kafka中的数据对应我在ods层中的一张表，然后我们可以通过sql对ods层进行操作，为dim，dwd层提供数据

注意：我们在构建dim层的维表并将维表的数据写入到hbase的时候，发现一个错误。就是写入到hbase的数据会变少。这可能是一个bug。

• 将维度表单独保存到一个hbae的命名空间(相当于库)中

  # 进入hbase shell
  habse shell
  # 创建命名空间
  create_namespace 'gma_dim'
  

1、地区维度表

将省份表和地区表合并成地区维度表(维度退化)
从kafka中读取数据合并两个表，将合并之后的地区维度表保存到hbae中

-- 1、在hbase中创建表
create 'gma_dim.dim_region','info'
 
-- 2、在flink实时数据仓库的dim层创建地区维度表
-- 创建hbase sink表
CREATE TABLE gma_dim.dim_hbase_region (
 pro_id BIGINT,
 info ROW<pro_name STRING,region_id BIGINT,region_name STRING,area_code STRING>,
 PRIMARY KEY (pro_id) NOT ENFORCED
) WITH (
 'connector' = 'hbase-1.4',
 'table-name' = 'gma_dim.dim_region',
 'zookeeper.quorum' = 'master:2181'
);
 
 
-- 3、编写flinksql 合并地区表和省份表，将数据保存到地区维度表中
insert into gma_dim.dim_hbase_region
select pro_id,ROW(pro_name,region_id,region_name,area_code) from (
select b.id as pro_id,b.name as pro_name,region_id,region_name,area_code from
gma_ods.ods_mysql_kafka_base_region /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as a
inner join 
gma_ods.ods_mysql_kafka_base_province /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as b
on
a.id=b.region_id
) as c
 
--4、到hbase中查看是否有数据
scan 'gma_dim.dim_region'

2、商品维度表

商品表、品类表、spu表、商品三级分类、商品二级分类、商品一级分类表退化为商品维度表

维度退化：将多个维度表退化成一个维度表

-- 1、在hbsae中创建商品维度表
create 'gma_dim.dim_item_info','info'
 
-- 2、在flink实时数据仓库中创建商品维度表
CREATE TABLE gma_dim.dim_hbase_item_info (
 sku_id bigint,
 info ROW<spu_id bigint, price decimal(10,0) , sku_name STRING, sku_desc STRING, weight  decimal(10,2), tm_id bigint, tm_name STRING, category3_id bigint, category3_name STRING, category2_id bigint, category2_name STRING, category1_id bigint, category1_name STRING, sku_default_img STRING, create_time TIMESTAMP(3) >,
 PRIMARY KEY (sku_id) NOT ENFORCED
) WITH (
 'connector' = 'hbase-1.4',
 'table-name' = 'gma_dim.dim_item_info',
 'zookeeper.quorum' = 'master:2181',
 'sink.buffer-flush.max-rows'='0'
);
 
 
--3、关联多张表得到商品维度表
insert into  gma_dim.dim_hbase_item_info 
select 
sku_id,ROW(spu_id ,  price,  sku_name,  sku_desc,  weight,  tm_id,  tm_name,  category3_id,  category3_name,  category2_id,  category2_name,  category1_id,  category1_name,  sku_default_img,  create_time) as info
from (
select 
a.id as sku_id,
a.spu_id,
a.price,
a.sku_name,
a.sku_desc,
a.weight,
a.tm_id,
b.tm_name,
a.category3_id,
c.name as category3_name,
d.id as category2_id,
d.name as category2_name,
e.id as category1_id,
e.name as category1_name,
a.sku_default_img,
a.create_time
from 
gma_ods.ods_mysql_kafka_sku_info 
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as a
inner join 
gma_ods.ods_mysql_kafka_base_trademark 
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as b
on a.tm_id=cast(b.tm_id as bigint)
inner join 
gma_ods.ods_mysql_kafka_base_category3
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as c
on a.category3_id=c.id
inner join 
gma_ods.ods_mysql_kafka_base_category2
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as d
on c.category2_id=d.id
inner join 
gma_ods.ods_mysql_kafka_base_category1
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as e
on d.category1_id=e.id
) as f;

3、用户维度表

--1、在hbase中创建用户维度表
create 'gma_dim.dim_user_info','info'
 
--2、在flink实时数据仓库中创建用户维度表
CREATE TABLE gma_dim.dim_hbase_user_info (
 user_id BIGINT,
 info ROW<login_name STRING , nick_name STRING , passwd STRING , name STRING , phone_num STRING , email STRING , head_img STRING , user_level STRING , birthday DATE , gender STRING , create_time TIMESTAMP(3)>,
 PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
 'connector' = 'hbase-1.4',
 'table-name' = 'gma_dim.dim_user_info',
 'zookeeper.quorum' = 'master:2181'
);
 
--3、编写flink sql读取ods层用户信息表将数据保存到hbae中构建用户维度表
insert into gma_dim.dim_hbase_user_info
select id as user_id,ROW(login_name,nick_name,passwd,name,phone_num,email,head_img,user_level,birthday,gender,create_time) from 
gma_ods.ods_mysql_kafka_user_info
/*+ OPTIONS('scan.startup.mode'='earliest-offset') */ as a
;