hive数据仓库构建ods层模型案例

ODS(operational data store) 操作数据存储层, 又叫贴源层, 通常只把业务系统的数据抽取到hdfs,不做任务处理. 分为离线和实时同步,离线这块用的组件一般是sqoop,kettle及大厂自研的数据同步软件. 实时这块通常是通过canel读取数据库操作日志写入kafka,使用flink读kafka写hdfs.

从增量和全量的角度来看分为以下三种:离线全量,离线增量,实时增量

当前只讨论离线的方式如何来构建hive的ods层模型

1.查询业务表数据
select * from t1;
id	name	createtime	updatetime
1	p1	2024-12-03 00:00:00.0	2024-12-03 00:00:00.0
2	p2	2024-12-03 00:00:00.0	2024-12-03 00:00:00.0
3	p3	2024-12-03 00:00:00.0	2024-12-03 00:00:00.0

2.建hive内表
CREATE TABLE test_db.t1(
  id string, 
  name string, 
  createtime string, 
  updatetime string
)
partitioned by (etl_date string)
row format delimited fields terminated by '|' 
STORED AS textfile
;

3.sqoop同步全量数据到hdfs -> 离线每天全量(每天同步业务系统全量数据到hdfs,在hive做先删后插):
${sqoop_home}/sqoop import \
-Dorg.apache.sqoop.splitter.allow_text_splitter=true \
--connect jdbc:mysql://10.22.133.144:2883/prdb \
--username yjx_dml \
--password 'test#123' \
--driver com.oceanbase.jdbc.Driver \
--fields-terminated-by '|' \
--null-string '\\N' \
--null-non-string '\\N' \
--hive-delims-replacement ' ' \
--outdir /tpdata/data/sqoopcode \
--delete-target-dir \
--fetch-size 200 \
--map-column-hive ID=String,NAME=String,CREATETIME=String,UPDATETIME=String \
--target-dir /user/hive/warehouse/test_db.db/t1/etl_date=20241204 \
--query "select replace(t.ID,'|',' '),replace(t.NAME,'|',' '),CREATETIME,UPDATETIME from dsg.t1 t where \$CONDITIONS" \
-m 1

4.查询hive表
msck repair table t1;
select * from t1;
+--------+----------+------------------------+------------------------+--------------+
| t1.id  | t1.name  |     t1.createtime      |     t1.updatetime      | t1.etl_date  |
+--------+----------+------------------------+------------------------+--------------+
| 1      | p1       | 2024-12-03 00:00:00.0  | 2024-12-03 00:00:00.0  | 20241204     |
| 2      | p2       | 2024-12-03 00:00:00.0  | 2024-12-03 00:00:00.0  | 20241204     |
| 3      | p3       | 2024-12-03 00:00:00.0  | 2024-12-03 00:00:00.0  | 20241204     |
+--------+----------+------------------------+------------------------+--------------+

5. 查询业务表数据(有一条更新,一条新增的数据)
select * from t1 order by updatetime;
id	name	createtime	updatetime
3	p3	2024-12-03 00:00:00.0	2024-12-03 00:00:00.0
2	p2	2024-12-03 00:00:00.0	2024-12-03 00:00:00.0
1	p11	2024-12-03 00:00:00.0	2024-12-04 00:00:00.0
4	p4	2024-12-03 00:00:00.0	2024-12-04 00:00:00.0

5.sqoop同步增量数据到hdfs -> 离线每天增量(每天同步业务系统增量数据到hdfs):
${sqoop_home}/sqoop import \
-Dorg.apache.sqoop.splitter.allow_text_splitter=true \
--connect jdbc:mysql://10.22.133.144:2883/prdb \
--username yjx_dml \
--password 'test#123' \
--driver com.oceanbase.jdbc.Driver \
--fields-terminated-by '|' \
--null-string '\\N' \
--null-non-string '\\N' \
--hive-delims-replacement ' ' \
--outdir /tpdata/data/sqoopcode \
--delete-target-dir \
--fetch-size 200 \
--map-column-hive ID=String,NAME=String,CREATETIME=String,UPDATETIME=String \
--target-dir /user/hive/warehouse/test_db.db/t1/etl_date=20241205 \
--query "select replace(t.ID,'|',' '),replace(t.NAME,'|',' '),CREATETIME,UPDATETIME from dsg.t1 t where date_format(updatetime,'yyyyMMdd') >'20241003' and \$CONDITIONS" \
-m 1

6. 增量数据合入全量表
set exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;

msck repair table test_db.t1;
INSERT overwrite table  test_db.t1 PARTITION(etl_date)
select id,name,createtime,updatetime from 
(
SELECT
row_number()over(PARTITION BY  id ORDER BY updatetime desc) rn 
,id,name,createtime,updatetime
from t1
) tt	
where rn=1
;
+-----+-------+------------------------+------------------------+
| id  | name  |       createtime       |       updatetime       |
+-----+-------+------------------------+------------------------+
| 1   | p11   | 2024-12-03 00:00:00.0  | 2024-12-04 00:00:00.0  |
| 2   | p2    | 2024-12-03 00:00:00.0  | 2024-12-03 00:00:00.0  |
| 3   | p3    | 2024-12-03 00:00:00.0  | 2024-12-03 00:00:00.0  |
| 4   | p4    | 2024-12-03 00:00:00.0  | 2024-12-04 00:00:00.0  |
+-----+-------+------------------------+------------------------+
4 rows selected (32.916 seconds)