一、引言
在上一篇博客中，我们通过 DataX 成功将 MySQL 业务库中的数据采集到了 Hive 的 ODS（贴源层）。此时的数据虽然进来了，但依然保持着业务系统原始的面貌：数据杂乱、含有无效值、缺乏全局规范，且直接查询性能极差。
要让数据真正产生价值，我们需要进行数仓建模。业界标准的数仓架构通常分为四层：ODS（贴源层） -> DWD（明细层） -> DWS（汇总层） -> ADS（应用层），同时辅以 DIM（维度层） 提供全局统一的视角。
本篇博客将带领大家完成地铁刷卡数据从 ODS 到 ADS 的全链路建模，并引入新一代列式数据库 ClickHouse (CK)，解决 Hive 无法支撑高并发、低延迟实时分析的痛点。
二、DWD层建模：数据清洗与事实提取
DWD（Data Warehouse Detail）层是数仓的基石，主要负责数据清洗、脱敏、规范化，构建业务过程的事实表。
2.1 地铁出入站事实表
业务痛点：ODS 层的数据包含了“巴士”等非地铁数据，且存在系统测试时间（如半夜0点-6点）的脏数据。同时，交易类型是“地铁入站”、“地铁出站”，不够规范。
解决方案：创建 DWD 出入站表，过滤无效数据，并将交易类型标准化为 I（进站）和 O（出站），同时按天和交易类型进行双分区。
-- 1. 创建DWD出入站表
CREATE EXTERNAL TABLE dwd.dwd_fact_szt_in_out_detail(
car_no STRING COMMENT '车牌号',
card_no STRING COMMENT '卡号',
close_date STRING COMMENT '结算日期',
company_name STRING COMMENT '公司名称',
conn_mark STRING COMMENT '联程标记',
deal_date STRING COMMENT '交易时间',
deal_money STRING COMMENT '交易金额',
deal_value STRING COMMENT '交易值',
equ_no STRING COMMENT '设备编码',
station STRING COMMENT '线路站点'
)
PARTITIONED BY (dt STRING COMMENT '按天分区', deal_type STRING COMMENT '交易类型')
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\u0001'
STORED AS ORC
LOCATION '/daas/dwd/dwd_fact_szt_in_out_detail';
-- 2. 数据清洗与装载
INSERT INTO TABLE dwd.dwd_fact_szt_in_out_detail PARTITION(dt='20180901', deal_type)
SELECT
car_no, card_no, close_date, company_name, conn_mark,
deal_date, deal_money, deal_value, equ_no, station,
-- 标准化交易类型
CASE WHEN deal_type='地铁入站' THEN 'I' ELSE 'O' END AS deal_type
FROM ods.ods_szt_data
WHERE dt='20180901'
AND deal_type != '巴士'
-- 过滤非营业时间的异常数据
AND date_format(deal_date,'HH:mm') > '06:14'
AND date_format(deal_date,'HH:mm') < '23:59';
2.2 地铁乘坐事实表
业务痛点：刷卡记录是“流水账”（一进一出两行数据），但分析通常需要“一次完整行程”（一行数据：包含进站、出站、耗时等）。
解决方案：利用 LAG 窗口函数，在按卡号分组、按时间排序的基础上，将上一行的“进站信息”拉取到当前“出站”行。
-- 1. 创建乘坐事实表
CREATE EXTERNAL TABLE dwd.dwd_take_subway_detail(
card_no STRING, car_no STRING, take_day STRING,
in_company_name STRING, out_company_name STRING,
in_station STRING, out_station STRING,
in_time STRING, out_time STRING,
in_equ_no STRING, out_equ_no STRING,
conn_mark INT, deal_money DOUBLE, take_time DOUBLE
)
PARTITIONED BY (dt STRING)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\u0001'
STORED AS ORC;
-- 2. 提取完整行程
INSERT OVERWRITE TABLE dwd.dwd_take_subway_detail PARTITION(dt='20180901')
SELECT
card_no, car_no,
date_format(in_deal_date,'yyyy-MM-dd') AS take_day,
in_company_name, out_company_name, in_station, out_station,
in_deal_date AS in_time, out_deal_date AS out_time,
in_equ_no, out_equ_no,
IF(in_company_name = out_company_name, 1, 0) AS conn_mark,
deal_money,
-- 计算乘车时长（分钟）
ROUND((unix_timestamp(out_deal_date) - unix_timestamp(in_deal_date)) / 60, 2) AS take_time
FROM (
SELECT
car_no, card_no,
deal_date AS out_deal_date, station AS out_station,
company_name AS out_company_name, equ_no AS out_equ_no,
deal_type, deal_money,
-- 核心：获取上一条进站记录
LAG(company_name, 1) OVER(PARTITION BY card_no ORDER BY deal_date) AS in_company_name,
LAG(station, 1) OVER(PARTITION BY card_no ORDER BY deal_date) AS in_station,
LAG(deal_date, 1) OVER(PARTITION BY card_no ORDER BY deal_date) AS in_deal_date,
LAG(equ_no, 1) OVER(PARTITION BY card_no ORDER BY deal_date) AS in_equ_no,
LAG(deal_type, 1) OVER(PARTITION BY card_no ORDER BY deal_date) AS lag_deal_type
FROM dwd.dwd_fact_szt_in_out_detail
WHERE dt='20180901'
) AS a
WHERE lag_deal_type = 'I' AND deal_type = 'O'
AND out_station != '""' AND in_station != '""';
三、DIM层与DWS层建模：构建全局宽表
3.1 DIM层：用户维度脱敏
维度层（DIM）存放公共的维度数据。出于隐私保护，我们需要对身份证、姓名进行 MD5脱敏。
-- 创建并处理用户维度表
INSERT OVERWRITE TABLE dim.dim_szt_user PARTITION(dt='20180901')
SELECT
md5(id) AS id, -- 身份证脱敏
md5(name) AS name, -- 姓名脱敏
card_no, age, sex
FROM ods.ods_szt_user
WHERE dt='20180901';
3.2 DWS层：宽表构建
DWS（汇总层）的核心任务是宽表化——将事实表与维度表关联，减少下游 ADS 层的重复关联计算。
-- 将乘坐事实表与用户维度表进行 Join
INSERT OVERWRITE TABLE dws.dws_take_subway_detail PARTITION(dt='20180901')
SELECT
a.card_no, a.car_no,
year(a.take_day) AS take_year, month(a.take_day) AS take_month, a.take_day,
a.in_company_name, a.out_company_name, a.in_station, a.out_station,
a.in_time, a.out_time, a.in_equ_no, a.out_equ_no,
b.age, b.sex,
a.conn_mark, a.deal_money, a.take_time
FROM (
SELECT * FROM dwd.dwd_take_subway_detail WHERE dt='20180901'
) AS a
JOIN (
SELECT * FROM dim.dim_szt_user WHERE dt='20180901'
) AS b
ON a.card_no = b.card_no;
四、ADS层建模：数据赋能业务应用
ADS 层直接面向业务报表，数据极其精简。基于 DWS 宽表，我们可以轻松算出各类排行榜。
4.1 指标1：进站人次排行
SELECT station, COUNT(1) AS flow
FROM dwd.dwd_fact_szt_in_out_detail
WHERE dt='20180901' AND deal_type = 'I'
GROUP BY station
ORDER BY flow DESC
LIMIT 100;
4.2 指标2：热门乘车区间排行榜
逻辑：无论从A到B还是B到A，都算同一个区间。使用 CASE WHEN 统一格式（如让字典序小的站点在前）。
SELECT
CASE
WHEN in_station > out_station THEN concat(in_station, '-', out_station)
ELSE concat(out_station, '-', in_station)
END AS station_pair,
COUNT(1) AS flow
FROM dwd.dwd_take_subway_detail
WHERE dt='20180901'
GROUP BY
CASE
WHEN in_station > out_station THEN concat(in_station, '-', out_station)
ELSE concat(out_station, '-', in_station)
END
ORDER BY flow DESC
LIMIT 100;
4.3 指标3：单程直达平均耗时
这里引入了 OneData模型思想：统计周期（每天） + 粒度（线路、性别） + 限定（直达） + 原子指标（耗时平均值）。
-- 由于维度组合太多，如果每个组合都建表会导致表爆炸，所以推荐建视图
CREATE VIEW ads.ads_line_single_ride_average_time_day_top AS
SELECT
in_company_name, sex,
ROUND(AVG(take_time), 2) AS avg_take_time
FROM dws.dws_take_subway_detail
WHERE dt='20180901' AND conn_mark = 1 -- 单程直达限定
GROUP BY in_company_name, sex
ORDER BY avg_take_time DESC;
五、 ClickHouse
5.1 为什么需要 ClickHouse？
随着业务发展，高层领导需要在大屏上实时刷新报表。此时，传统的架构遇到了瓶颈：
-
MySQL：适合事务（OLTP），做千万级数据的分组聚合极慢。
-
Hive：适合海量离线分析，但底层是 MapReduce/Spark，延迟在分钟级以上，无法满足“秒级响应”。
ClickHouse (CK) 是俄罗斯 Yandex 开源的一款列式存储数据库（OLAP），以极致的查询性能著称，非常适合做实时大屏和交互式分析。
5.2 ClickHouse 极速搭建
# 1. 安装 RPM 包
rpm -ivh clickhouse-common-static-*.rpm
rpm -ivh clickhouse-server-*.rpm
rpm -ivh clickhouse-client-*.rpm
# 2. 修改配置 (端口冲突与外部访问)
cd /etc/clickhouse-server/
vim config.xml
# 修改 tcp_port 为 9001（避免与HDFS冲突）
# 开启 <listen_host>0.0.0.0</listen_host> 允许外部访问
# 3. 启动服务
systemctl start clickhouse-server.service
# 4. 客户端连接测试
clickhouse-client --port 9001 -m
5.3 建表与引擎（MergeTree）
CK 的核心是 MergeTree 引擎，建表时必须指定 Order By（排序键，相当于主键索引）。
CREATE DATABASE IF NOT EXISTS dws;
CREATE TABLE dws.dws_take_subway_detail(
card_no String,
car_no String,
take_year String,
take_month String,
take_day String,
in_company_name String,
out_company_name String,
in_station String,
out_station String,
in_time String,
out_time String,
in_equ_no String,
out_equ_no String,
age UInt32,
sex String,
conn_mark UInt32,
deal_money Float32,
take_time Float32,
dt String
)
ENGINE = MergeTree()
PARTITION BY dt
ORDER BY (card_no); -- 根据经常查询的字段设置排序键
六、DataX 将 Hive 数据写入 CK
宽表在 Hive 计算完毕后，如何同步到 ClickHouse 供大屏展示？依然可以使用 DataX。
这次我们将 Reader 设置为 hdfsreader（读取 Hive ORC 文件），Writer 设置为 clickhousewriter。
DataX 核心配置 hdfs_to_ck.json：
{
"job": {
"content": [
{
"reader": {
"name": "hdfsreader",
"parameter": {
"path": "/daas/dws/dws_take_subway_detail/dt=20180901/*",
"defaultFS": "hdfs://master:9000",
"fileType": "orc",
"column": [
{"index": 0, "type": "String"},
{"index": 1, "type": "String"},
// ... 省略中间字段 ...
{"index": 17, "type": "DOUBLE"},
{"type": "string", "value": "20180901"} // 补充常量 dt 字段
],
"fieldDelimiter": "\u0001"
}
},
"writer": {
"name": "clickhousewriter",
"parameter": {
"username": "default",
"password": "123456",
"column": [
"card_no", "car_no", "take_year", "take_month", "take_day",
"in_company_name", "out_company_name", "in_station", "out_station",
"in_time", "out_time", "in_equ_no", "out_equ_no",
"age", "sex", "conn_mark", "deal_money", "take_time", "dt"
],
"connection": [
{
"jdbcUrl": "jdbc:clickhouse://master:8123/dws",
"table": ["dws_take_subway_detail"]
}
],
"preSql": ["ALTER TABLE dws.dws_take_subway_detail DROP PARTITION '20180901'"],
"writeMode": "insert"
}
}
}
]
}
}
注意：preSql 是一个非常重要的幂等性设计，在同步前先删除 CK 中的旧分区，防止重跑导致数据翻倍。
七、总结
数据仓库系列博客完整走完了一个企业级数仓从建仓 -> 建模 -> 应用 -> 实时加速的标准生命周期。
| 层次/组件 | 核心职责 | 关键技术点 |
|---|---|---|
| DWD 明细层 | 数据清洗、事实提取 | 过滤脏数据、LAG 窗口函数打平行程 |
| DIM 维度层 | 统一维度、隐私保护 | MD5 脱敏 |
| DWS 汇总层 | 宽表构建、降维提效 | 事实表与维度表 JOIN |
| ADS 应用层 | 业务指标计算 | CASE WHEN 标准化、视图封装 |
| ClickHouse | 实时OLAP加速分析 | MergeTree 引擎、列式存储 |
| DataX | 跨数据源同步 | hdfsreader -> clickhousewriter |
架构演进心得：
-
为什么要分层？为了解耦。底层数据变动不影响上层应用，同时宽表化极大提升了 ADS 层的查询速度。
-
为什么要用 ClickHouse？Hive 负责“海量离线 T+1 计算”，CK 负责“秒级实时查询”。两者结合，才能兼顾成本与性能。
-
数据同步幂等性：在使用 DataX 跑批时，一定要利用
preSql删除旧分区或使用hive -e "insert overwrite"机制，确保任务重复执行不产生脏数据。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/2301_77064095/article/details/169463990
