hive 分层设计文档
分层本质：解耦、复用、易排查、稳口径、提性能
通用分层链路：ODS原始层 → DWD明细清洗层 → DWS汇总宽表层 → ADS指标应用层
一、隔离原始数据，保护源头，支持数据回溯
- ODS 原样同步业务库数据，不做任何清洗、不修改字段，完整留存原始快照；
- 下游 DWD/DWS/ADS 只读取 ODS，不会直接操作原始数据源；
- 好处：
- 业务系统字段改动、脏数据、乱码、缺失值全部隔离在 ODS，不会污染上层指标；
- 报表数据出错时，可拿 ODS 原始数据重跑、回溯修复，不会丢失原始记录；
- 禁止业务人员直接访问原始业务库，减轻业务数据库压力。
举例：新发地源头数据avg_price存在 “未知、空字符串、负数” 脏数据，只在 DWD 统一过滤清洗，ODS 完整保留原始脏数据用于核对。
二、分层拆分加工逻辑，复杂问题简单化，方便排错
不分层的弊端：一张 SQL 从原始表直接算出最终报表，清洗、去重、关联、聚合全部写在一段，逻辑臃肿，出问题无从定位。
分层拆解每一层只做一件事：
- ODS：同步数据；
- DWD：清洗脏数据、统一字段类型、规范日期、去重；
- DWS：按日期 / 分类轻度汇总，生成宽表；
- ADS：业务指标高度聚合（涨跌、价格分布、品类统计）。
排错链路（天然数据血缘）
ADS 报表数据异常 → 查 DWS 汇总逻辑 → 查 DWD 明细脏数据 → 核对 ODS 原始源头
逐层定位，不用通读几百行超长 SQL，排查效率提升 80%。
三、公共计算只做一次，减少重复开发、节省集群资源
不分层场景：市场报表、价格分布、品类统计三张报表，每个人都从 ODS 重新清洗、转换价格、分类，重复跑相同逻辑
分层方案：
- DWD 统一清洗一次干净商品明细；
- DWS 按天 / 分类汇总一次公共指标；
- 所有 ADS 报表直接复用 DWS 结果，不再重复清洗、重复聚合。
价值：大幅减少 MapReduce 任务数量，降低 HDFS 读写压力，解决重复计算资源浪费问题。
四、统一数据口径，杜绝 “数据打架”
最核心业务价值：一数一源
如果每个报表自己写清洗规则：
-
A 报表把价格 < 0 过滤掉；B 报表保留负数；
-
A 把 “未知产地” 剔除，B 不剔除；
最终两张报表商品总量、均价对不上，业务质疑数据不准。
分层统一规则：
所有清洗、过滤、区间划分逻辑只在 DWD 层定义一次，全平台所有指标统一复用同一套标准，口径完全一致。
五、分层适配查询性能，分层优化存储
- DWD：保留最细明细，适合明细查询、数据分析；
- DWS：提前按主题聚合宽表，减少多表 join，日常统计查询速度极快；
- ADS：高度聚合，单表直接给大屏、BI 报表使用，几乎无计算开销；
- 存储分层优化：ODS 可用 Text 存储；DWD/DWS/ADS 统一 ORC+SNAPPY 压缩，分层设置合并小文件参数，各司其职优化存储。
六、解耦变更，提高架构稳定性
- 源系统表结构变更（新增字段、改名）：只修改 ODS 同步脚本，DWD 做兼容转换，上层 DWS/ADS 不用改动；
- 业务需求调整指标计算逻辑：仅修改对应 ADS 层，明细、汇总层不受影响；
- 多层单向依赖（下层供上层，禁止反向依赖），一层改动不会连锁污染全链路，维护成本极低。
1、总体分层架构
┌─────────────────────────────────── ─┐
│ ads 应用层（可选） │
│ 面向业务输出：报表 / 大屏 / 仪表盘 / 接口数据 │
└─────────────────────────────────────────────────────────┘
▲
┌─────────────────────────────────────────────────────────┐
│ dws 汇总层（宽表层） │
│ 共性指标汇总：品类汇总 / 产地汇总 / 价格区间汇总 / 综合宽表 │
└─────────────────────────────────────────────────────────┘
▲
┌─────────────────────────────────────────────────────────┐
│ dwd 明细层（事实层） │
│ 清洗后的原子明细记录：二次过滤、类型转换、去重、补齐审计字段 │
└─────────────────────────────────────────────────────────┘
▲
┌─────────────────────────────────────────────────────────┐
│ ods 贴源层（操作数据层） │
│ 原始 CSV 原样映射到 Hive 外部表，保持与源系统一致 │
└─────────────────────────────────────────────────────────┘
▲
┌─────────────────────────────────────────────────────────┐
│ HDFS 原始数据层 │
│ /xinfadi/*.csv 每日采集 CSV 文件 │
└─────────────────────────────────────────────────────────┘
2、各层功能作用
| 层级 | 中文名 | 核心目标 | 表类型 | 数据保留策略 |
|---|---|---|---|---|
| ODS | 操作数据层 / 贴源层 | 原样保存源数据，便于回溯与重跑 | 外部表 (EXTERNAL) | 全量，长期保留 |
| DWD | 明细数据层 | 清洗、去重、标准化后的原子明细记录 | 内部表 (MANAGED) | 全量有效明细 |
| DWS | 汇总数据层 | 按主题聚合，形成共性指标宽表 | 内部表 (MANAGED) | 每日聚合快照 |
| ADS | 应用数据层 | 面向具体业务报表输出 | 内部表 / 视图 | 按需保留 |
2.1 ods 层：Operation Data Store
设计规则：
- 不做任何数据清洗，字段与源 csv一一对应
- 使用
EXTERNAL TABLE外部表，数据存放在 hdfs/xinfadi - 通过
tblproperties('skip.header.line.count'='1')自动跳过 csv 表头 - 价格类字段先用
string保存原始值，避免类型转换失败导致丢数 - 字段命名：英文驼峰（便于后续与 dwd 字段映射）
功能作用：
- 数据采集后第一站，数仓的"原始镜像"
- 后续任何重跑、问题追溯，都可从 ods 重新计算
- 保留数据原貌，不影响源系统或原始采集文件
2.2 dwd 层：Data Warehouse Detail
设计规则：
- 基于 ods 做清洗，输出"干净、可用的明细数据"
- 关键动作：
- 缺失值过滤：品名 / 分类 / 价格 / 发布时间 为空 → 删除
- 类型转换：价格
string → DOUBLE；日期统一格式YYYY-MM-DD - 逻辑校验：
最低价 ≤ 平均价 ≤ 最高价，且> 0 - 全局上下限：
0.1 ≤ 平均价 ≤ 2000 - 分组 IQR 去极端值：按
(一级分类, 品名)分组剔除 Q1-3·IQR 以外的异常值 - 去重：
(分类, 品名, 产地, 规格, 单位, 日期, 价格)组合键去重 - 补齐审计字段：
id（业务唯一标识）、create_time、update_time
存储策略：
- 内部表（
MANAGED TABLE），Hive 管理生命周期 stored as PARQUET+SNAPPY压缩，列式存储、高效查询
功能作用：
- 数仓中最核心的"事实明细"，所有后续统计的基础
- 分析师、业务人员直接查 dwd 就能拿到可信数据，不用再关心 ods 的脏数据
- 可以直接关联维度表（例如商品维度、产地维度、日期维度）做复杂分析
2.3 dws 层：Data Warehouse Summary
设计规则：
- 以 dwd 明细层 为唯一输入，按不同主题做聚合
- 每张 dws 表围绕"统计日期 + 聚合维度"形成宽表
- 每个宽表都应当是"同一粒度下的完整业务视角"，下游 ads / 报表无需再 join
- 存储：
PARQUET，便于 BI / Spark / Impala 等多种引擎高效读取 - 命名：
dws_<主题>_<时间粒度>，如dws_category_day（品类日汇总）
功能作用：
- 共性指标沉淀，避免重复计算
- 数据体量显著缩减（dwd 可能几十万行，dws 聚合后仅几千行）
- 对外提供稳定的数据接口，业务报表、大屏直接消费 DWS
2.4、数仓分层设计原则总结（可放报告正文）
- 一次清洗，多次复用：清洗工作集中在 dwd 层完成，dws 直接从 dwd 取数，避免各层重复做清洗。
- 数据可回溯：ods 保留原始数据，任意时刻发现清洗逻辑错误，都可以从 ods 重跑 dwd → DWS，不影响业务。
- 逐层收敛：ODS（百万行）→ DWD（十万行级）→ DWS（万/千行级）→ ADS（报表级），数据越上层量越小，越贴近业务。
- 宽表思想：dws 每张表都把一个主题"做宽"，把相关指标放在同一张表，下游查询不需要 join，性能和可维护性都更好。
- 统一命名规范：
<层前缀>_<主题>_<粒度>，如dws_category_day、dwd_xinfadi_data，任何人看到表名就能猜出它是什么、来自哪层。 - 审计字段：dwd 及以上所有表都带
id / create_time / update_time，支持数据血缘追溯与增量更新。 - 外部表 vs 内部表：ods 用外部表保护源数据；DWD/dws 用内部表（PARQUET+SNAPPY）享受 Hive 自动管理 + 列式存储性能。
2.5、数据流向示意（配图用）
爬虫采集 (py1)
│
▼
data/YYYY-MM-DD.csv (本地文件)
│
▼ scp / sftp + hdfs dfs -put
▼
HDFS /xinfadi/*.csv (原始文件系统)
│
▼ CREATE EXTERNAL TABLE LOCATION '/xinfadi'
▼
ods_xinfadi_price (外部表，string 原始字段)
│
▼ 清洗：类型转换 / 去重 / 异常值过滤 / 去重 / 补 ID
▼
dwd_xinfadi_data (内部表，PARQUET，原子明细)
│
▼ 按主题聚合
▼
┌─────────────┬──────────────┬───────────────┬──────────────┐
│ 品类汇总表 │ │ 价格区间汇总表 │ 综合宽表 │
│ dws_category │ │ dws_price_range│ dws_overall │
│ _day │ │ _day │ _day │
└─────────────┴───────────────┴─────────────────┴───────────────┘
│
▼ (可选) 面向 BI / 报表 / 大屏
▼
ads 应用层：报表表 / 接口表 / 大屏数据
3、ods 层：外部表
-- 1. 删除旧表（外部表，仅删元数据，HDFS 上的 CSV 不会丢）
drop database if exists xinfadi cascade;
-- 创建数据库（可选，分类管理）
create database if not exists xinfadi_db comment '新发地农产品价格数据仓库';
create database if not exists xinfadi comment '新发地农产品价格数据仓库';
-- 进入数据库
use xinfadi_db;
show databases;
show tables;
-- 创建外部表 location 指向你的HDFS文件目录
create external table if not exists ods_xinfadi_data (
prod_id string comment 'id',
prod_cat string comment '一级分类',
prod_pcat string comment '二级分类',
prod_name string comment '品名',
high_price string comment '最高价',
low_price string comment '最低价',
avg_price string comment '平均价',
spec_info string comment '规格',
place string comment '产地',
unit_info string comment '单位',
pub_date string comment '发布时间'
)
comment '原始csv ods 层，未做任何数据清洗'
row format delimited
fields terminated by ',' -- 文本文件每行用逗号切分各个字段 csv逗号分隔
stored as textfile -- 指定 Hive 表底层存储格式为纯文本文件
location '/xinfadi/' -- 指定 hdfs 数据存放目录
tblproperties (
"skip.header.line.count"="1",
"serialization.encoding" = "UTF-8"
); -- 忽略表头第一行
-- 3. 校验
desc ods_xinfadi_data;
select * from ods_xinfadi_data where pub_date ='2026-06-22 00:00:00' limit 5
select * from ods_xinfadi_data limit 5;
select count(*) from ods_xinfadi_data;
4、dwd 层：明细层内部表（清洗）
```mysql -- ===================================================== -- dwd 层：清洗后的明细数据（内部表，PARQUET 压缩存储） -- 做什么： -- 1) 删除关键字段缺失记录 -- 2) cast(字段 as 数据类型) 强制字段类型转换 价格 string -> decimal(总长度, 小数位数) ， id string-> bigint ，日期转 id string-> date -- 3) 过滤异常值（价格非正值 / 逻辑矛盾 / 极端值） -- 4) current_timestamp 是时间戳 增加 create_time / update_time 审计字段 -- 5) distinct 去重 -- 6) trim 去除字段前后的空格、制表符，避免空格干扰正则判断。 -- 7) rlike '^[0-9]+$' 正则匹配规则 ^开头字符串结尾 只能是数字 0~9，+：数字至少出现 1 次 匹配小数点 .（Hive 里反斜杠要转义
-- 价格字段 两处正则 + 转换作用完全不一样，执行阶段不同 -- 第一次 1. WHERE 里的正则 + cast（前置过滤，筛掉整条脏行） cast 只判断 不完成类型转换 -- 第二次 2. SELECT 子查询的 case 正则 + cast（字段标准化）cast 判断+成类型转换 -- 建议两个都要写 只走select 会大量无效数据参与计算，拖慢 SQL 执行速度； 只where 又无法类型转换
-- =====================================================
drop table if exists dwd_xinfadi_data;
-- dwd 新发地清洗明细数据表 create table if not exists dwd_xinfadi_data ( prod_id bigint comment 'ID，纯数字有效，非法为null', prod_cat string comment '一级分类，已去首尾空格', prod_pcat string comment '二级分类，已去首尾空格', prod_name string comment '品名，已去首尾空格', high_price decimal(12,2) comment '最高价，合法正数数值', low_price decimal(12,2) comment '最低价，合法正数数值', avg_price decimal(12,2) comment '平均价，合法正数数值', spec_info string comment '规格，已去首尾空格', place string comment '产地，已去首尾空格', unit_info string comment '单位，已去首尾空格', pub_date date comment '发布日期，标准YYYY-MM-DD格式', create_time timestamp comment '数据入库创建时间', update_time timestamp comment '数据更新时间' ) comment '新发地商品价格清洗明细层DWD表，过滤脏数据、标准化字段类型' stored as orc tblproperties ( "orc.compress" = "SNAPPY", "comment" = "清洗后明细，过滤空值、非法价格、价格逻辑错乱、非法日期" );
insert overwrite table dwd_xinfadi_data select t.prod_id, t.prod_cat, t.prod_pcat, t.prod_name, t.high_price, t.low_price, t.avg_price, t.spec_info, t.place, t.unit_info, t.pub_date, current_timestamp as create_time, current_timestamp as update_time from ( select distinct -- 商品ID 如果 prod_id 是纯数字字符串，把字符串强转为大整数 bigint类型 case when trim(prod_id) rlike '^[0-9]+$' then cast(trim(prod_id) as bigint) else null end as prod_id, trim(prod_cat) as prod_cat, trim(prod_pcat) as prod_pcat, trim(prod_name) as prod_name, trim(spec_info) as spec_info, trim(place) as place, trim(unit_info) as unit_info,
-- 最高价
case when trim(high_price) rlike '^[0-9]+(\\.[0-9]+)?$'
then cast(trim(high_price) as decimal(12,2))
else null end as high_price,
-- 最低价
case when trim(low_price) rlike '^[0-9]+(\\.[0-9]+)?$'
then cast(trim(low_price) as decimal(12,2))
else null end as low_price,
-- 平均价
case when trim(avg_price) rlike '^[0-9]+(\\.[0-9]+)?$'
then cast(trim(avg_price) as decimal(12,2))
else null end as avg_price,
-- 发布日期：兼容 2026-06-22 / 2026-06-22 00:00:00
case when trim(pub_date) rlike '^[0-9]{4}-[0-9]{2}-[0-9]{2}( [0-9]{2}:[0-9]{2}:[0-9]{2})?$'
then to_date(trim(pub_date))
else null end as pub_date
from ods_xinfadi_data
where
trim(prod_name) is not null and trim(prod_name) != ''
and trim(prod_cat) is not null and trim(prod_cat) != ''
and trim(prod_id) is not null and trim(prod_id) != ''
and trim(pub_date) is not null and trim(pub_date) != ''
and trim(avg_price) rlike '^[0-9]+(\\.[0-9]+)?$'
and cast(trim(avg_price) as decimal(12,2)) > 0
and cast(trim(avg_price) as decimal(12,2)) <= 100000
and trim(high_price) rlike '^[0-9]+(\\.[0-9]+)?$'
and cast(trim(high_price) as decimal(12,2)) > 0
and trim(low_price) rlike '^[0-9]+(\\.[0-9]+)?$'
and cast(trim(low_price) as decimal(12,2)) > 0
and cast(trim(low_price) as decimal(12,2)) <= CAST(trim(avg_price) as decimal(12,2))
and cast(trim(avg_price) as decimal(12,2)) <= CAST(trim(high_price) as decimal(12,2))
-- 修复日期正则，兼容带时分秒
and trim(pub_date) rlike '^[0-9]{4}-[0-9]{2}-[0-9]{2}( [0-9]{2}:[0-9]{2}:[0-9]{2})?$'
) t;
select * from dwd_xinfadi_data where pub_date ='2026-06-22' limit 5;
select * from ods_xinfadi_data where pub_date ='2026-06-22 00:00:00' limit 5
dwd_xinfadi_data
-- 看看清洗掉了多少条 select (select count() from ods_xinfadi_price) as ods_cnt, (select count() from dwd_xinfadi_data) as dwd_cnt;
```
5、dwd 层 基础指标查询
5.1 dwd 单表查询
--
-- 查 dwd 前 20 条
select * from dwd_xinfadi_data limit 20;
-- 查所有一级分类
select distinct prod_cat from dwd_xinfadi_data;
-- 日期覆盖范围
select min(pub_date_std) as min_date,
max(pub_date_std) as max_date,
count(distinct pub_date_std) as days
from dwd_xinfadi_data;
select count(*) as dwd_total from dwd_xinfadi_data;
5.2 dwd 条件筛选 (where)
select * from dwd_xinfadi_data where pub_date ='2026-06-22' limit 5;
-- 只看蔬菜类
select prod_name, avg_price, place, pub_date_std
from dwd_xinfadi_data
where prod_cat = '蔬菜'
and pub_date_std = '2026-06-16'
limit 20;
-- 平均价 > 50 的高档商品
select prod_name, prod_cat, avg_price, place
from dwd_xinfadi_data
where avg_price > 50
order by avg_price desc
limit 20;
-- 白菜相关商品
select prod_name, high_price, low_price, avg_price, place
from dwd_xinfadi_data
where prod_name LIKE '%白菜%'
order by avg_price;
-- 多条件：蔬菜 and 产地山东
select prod_name, avg_price, place, pub_date_std
from dwd_xinfadi_data
where prod_cat = '蔬菜'
and place LIKE '%山东%'
and avg_price between 1 and 10
limit 20;
5.3 dwd group by 分组统计
-- 按一级分类统计商品数 & 平均价
select
prod_cat,
count(*) as cnt,
round(avg(avg_price), 2) as avg_price
from dwd_xinfadi_data
group by prod_cat
order by cnt desc;
-- 按 (一级分类, 二级分类) 统计
select
prod_cat, prod_pcat,
count(*) as cnt,
round(avg(avg_price), 2) as avg_price,
round(max(avg_price), 2) as max_price
from dwd_xinfadi_data
group by prod_cat, prod_pcat
order by prod_cat, avg_price desc;
-- 按日统计：每天记录数
select
pub_date_std,
count(*) as total_rows,
count(distinct prod_name) as product_kinds,
round(avg(avg_price), 2) as avg_price
from dwd_xinfadi_data
group by pub_date_std
order by pub_date_std;
-- having：筛选平均价超过 20 的分类
select
prod_cat,
count(*) as cnt,
round(avg(avg_price), 2) as avg_price
from dwd_xinfadi_data
group by prod_cat
having avg(avg_price) > 20
order by avg_price desc;
5.4 dwd order by 排序
-- 按平均价降序 Top20（最贵商品）
select prod_name, prod_cat, place, avg_price, pub_date_std
from dwd_xinfadi_data
order by avg_price desc
limit 20;
-- 按平均价升序 Top20（最便宜商品）
select prod_name, prod_cat, place, avg_price
from dwd_xinfadi_data
order by avg_price asc
limit 20;
-- 多字段排序：先按分类，再按价格
select prod_cat, prod_name, avg_price
from dwd_xinfadi_data
order by prod_cat, avg_price desc;
5.5 dwd 基础指标计算
-- ===== 指标 1：各一级分类下 Top5 最贵商品（窗口函数）=====
-- Hive 窗口函数，作用：按商品大类分组，每组内部按均价从高到低排序，给每条数据生成组内自增序号。
-- over() 窗口函数
-- row_number () 无并列：就算数值一样，序号也依次递增（1、2、3）
-- dense_rank 规则：相同数值 → 同名次 下一名名次不跳号（1、1、2、3… 连续）
-- rank () 跳号：相同数值同名次，后面直接跳过占用的名次（1、1、3）
给分组排序后的行分配连续序号：1、2、3、4、5……
select first_category, product_name, avg_price
from (
select
prod_cat as first_category,
prod_name as product_name,
avg_price,
row_number() over(partition by prod_cat order by avg_price desc) as rn
from dwd_xinfadi_data
) t
where rn <= 5;
-- ===== 指标 2：每日价格波动（最高价 vs 最低价）=====
select
pub_date_std,
prod_cat,
round(avg(high_price - low_price), 2) as avg_price_diff,
round(max(high_price - low_price), 2) as max_price_diff
from dwd_xinfadi_data
group by pub_date_std, prod_cat
order by pub_date_std, prod_cat;
-- ===== 指标 3：某品类价格趋势（按日期分组）=====
select
pub_date_std,
round(avg(avg_price), 2) as daily_avg
from dwd_xinfadi_data
where prod_name LIKE '%白菜%'
group by pub_date_std
order by pub_date_std;
-- ===== 指标 4：各产地商品数量排名 =====
select
place,
count(*) as product_cnt,
dense_rank() over(order by count(*) desc) as rnk
from dwd_xinfadi_data
where place is not null and trim(place) != '未知'
group by place
order by product_cnt desc
limit 20;
-- ===== 指标 5：价格带分布 =====
-- 第一步 group by 先分组
-- 第二步 select 里的 case 生成别名
select
case
when avg_price < 5 then '0-5元'
when avg_price < 10 then '5-10元'
when avg_price < 20 then '10-20元'
when avg_price < 50 then '20-50元'
when avg_price < 100 then '50-100元'
else '100元以上'
end as price_band,
count(*) as cnt
from dwd_xinfadi_data
group by
case
when avg_price < 5 then '0-5元'
when avg_price < 10 then '5-10元'
when avg_price < 20 then '10-20元'
when avg_price < 50 then '20-50元'
when avg_price < 100 then '50-100元'
else '100元以上'
end
order by cnt desc;
-- 相同结果不同写法
select
price_band,
count(*) as cnt
from (
select
case
when avg_price < 5 then '0-5元'
when avg_price < 10 then '5-10元'
when avg_price < 20 then '10-20元'
when avg_price < 50 then '20-50元'
when avg_price < 100 then '50-100元'
else '100元以上'
end as price_band
from dwd_xinfadi_data
) t
group by price_band
order by cnt desc;
5、dws 层：各类汇总宽表
-- stored as orc -Hive 表底层文件存储格式为 ORC 列式存储文件。 查询速度快 自带索引
-- tblproperties 表配置额外属性 ORC 文件开启 SNAPPY 压缩算法 压缩速度快、解压开销极小
-- hive.exec.orc.split.strategy" = "BI" ORC 文件切片拆分策略
-- dws_product_info（商品信息表：商品维度表）
create table if not exists dws_product_info (
prod_cat string comment '一级分类',
prod_pcat string comment '二级分类',
prod_name string comment '品名'
)
comment 'DWS层-新发地商品维度信息表，去重后的商品基础维度'
stored as orc -- Hive 表底层文件存储格式为 ORC 列式存储文件。
tblproperties (
"orc.compress" = "SNAPPY",
"hive.exec.orc.split.strategy" = "BI"
);
-- 全量覆盖维度数据
insert overwrite table dws_product_info
select distinct
prod_cat,
prod_pcat,
prod_name
from dwd_xinfadi_data
where
prod_name is not null
and trim(prod_name) != '';
desc dws_product_info;
-- 检查
select * from dws_product_info order by `prod_cat`, `prod_pcat` limit 20;
-- 按一级分类统计商品数
select
prod_cat as cat,
count(*) as cnt
from dws_product_info group by prod_cat order by cnt desc;
-- 看看"蔬菜"分类下有哪些商品
select * from dws_product_info where prod_cat = '蔬菜'
order by prod_pcat, prod_name limit 30;
-- dws_xinfadi_price 品类汇总表
table if not exists dws_xinfadi_price (
pub_date date comment '统计日期',
prod_cat string comment '一级分类',
prod_name string comment '商品名',
record_count int comment '当天记录条数',
avg_price decimal(10,2) comment '平均价',
min_low_price decimal(10,2) comment '最低的最低价',
max_high_price decimal(10,2) comment '最高的最高价',
avg_price_range decimal(10,2) comment '平均价格浮动',
place_count int comment '不同产地数量'
)
comment 'DWS层-新发地商品每日价格汇总表'
stored as orc
tblproperties (
"orc.compress" = "SNAPPY"
);
insert overwrite table dws_xinfadi_price
select
t.pub_date,
t.prod_cat,
t.prod_name,
t.record_count,
t.avg_price,
t.min_low_price,
t.max_high_price,
t.avg_price_range,
t.place_count
from (
select
pub_date,
prod_cat,
prod_name,
count(*) as record_count,
round(avg(avg_price), 2) as avg_price,
round(min(low_price), 2) as min_low_price,
round(max(high_price), 2) as max_high_price,
round(avg(high_price - low_price), 2) as avg_price_range,
count(distinct place) as place_count
from dwd_xinfadi_data
where pub_date is not null
group by pub_date, prod_cat, prod_name
) t;
desc dws_xinfadi_price;
select * from dws_xinfadi_price limit 5;
-- 看 2026-06-17 这一天平均价最高的 5 件商品
select * from dws_xinfadi_price where pub_date='2026-06-15' order by avg_price desc limit 5
-- 1) 商品名里包含"肉"的所有记录
select * from dws_xinfadi_price where prod_name LIKE '%肉%' limit 5;
5.1 dws_price_range_day（价格区间每日汇总表）
create table if not exists dws_price_range_day (
pub_date date comment '统计日期',
price_range string comment '价格区间，如 [0-5)',
record_count int comment '该区间内记录条数',
avg_price decimal(10,2) comment '该区间平均价'
)
comment 'DWS层-价格区间每日汇总表'
stored as orc
tblproperties (
"orc.compress" = "SNAPPY"
);
insert overwrite table dws_price_range_day
select
t.pub_date,
t.price_range,
t.record_count,
t.avg_price
from (
select
pub_date,
price_range,
count(*) as record_count,
round(avg(avg_price), 2) as avg_price
from (
select
pub_date,
avg_price,
case
when avg_price < 5 then '[0-5)'
when avg_price < 10 then '[5-10)'
when avg_price < 20 then '[10-20)'
when avg_price < 50 then '[20-50)'
when avg_price < 100 then '[50-100)'
else '[100+)'
end as price_range
from dwd_xinfadi_data
where pub_date is not null and avg_price is not null
) tmp
group by pub_date, price_range
) t;
-- 检查
select count(*) cont from dws_price_range_day;
select * from dws_price_range_day group by pub_date order by pub_date desc limit 6;
-- 看最近一天的价格区间分布
select pub_date, price_range, record_count, avg_price
from dws_price_range_day
where pub_date = (select max(pub_date) from dws_price_range_day)
order by price_range;
5.2 dws_overall_day（综合每日二级分类汇总表）
-- 按「日期 + 一级分类 + 二级分类」分组的统计 SQL
create table if not exists dws_overall_day (
pub_date date comment '统计日期',
prod_cat string comment '一级分类',
prod_pcat string comment '二级分类',
record_count int comment '记录条数',
product_count int comment '不同商品数',
avg_price decimal(10,2) comment '平均价',
min_price decimal(10,2) comment '最低价',
max_price decimal(10,2) comment '最高价'
)
comment 'DWS层-综合每日汇总表（按日×一级×二级分类）'
stored as orc
tblproperties (
"orc.compress" = "SNAPPY"
);
insert overwrite table dws_overall_day
select
t.pub_date,
t.prod_cat,
t.prod_pcat,
t.record_count,
t.product_count,
t.avg_price,
t.min_price,
t.max_price
from (
select
pub_date,
prod_cat,
prod_pcat,
count(*) as record_count,
count(distinct prod_name) as product_count,
round(avg(avg_price), 2) as avg_price,
round(min(avg_price), 2) as min_price,
round(max(avg_price), 2) as max_price
from dwd_xinfadi_data
group by pub_date, prod_cat, prod_pcat
) t;
-- 检查
select pub_date,prod_cat,prod_pcat,record_count,product_count from dws_overall_day limit 10;
select * from dws_overall_day limit 10;
select count(*) cont from dws_overall_day;
-- 看最近一天的一、二级分类汇总
select pub_date, prod_cat, prod_pcat,
record_count, product_count, avg_price, min_price, max_price
from dws_overall_day
where pub_date = (select max(pub_date) from dws_overall_day)
order by record_count desc
limit 30;
5.3 dws_15day_summary（15日综合指标表）
每天使用 INSERT OVERWRITE 数据是清空后 才 插入的 不用担心日期问题
-- datediff(current_date(), pub_date) Hive 日期差值函数：返回 日期 1 − 日期 2 相差的天数
drop table if exists dws_15day_summary;
truncate table dws_15day_summary;
create table if not exists dws_15day_summary (
prod_cat string comment '一级分类',
prod_pcat string comment '二级分类',
prod_name string comment '品名',
d_14 decimal(12,2) comment '往前第14天均价',
d_13 decimal(12,2) comment '往前第13天均价',
d_12 decimal(12,2) comment '往前第12天均价',
d_11 decimal(12,2) comment '往前第11天均价',
d_10 decimal(12,2) comment '往前第10天均价',
d_9 decimal(12,2) comment '往前第9天均价',
d_8 decimal(12,2) comment '往前第8天均价',
d_7 decimal(12,2) comment '往前第7天均价',
d_6 decimal(12,2) comment '往前第6天均价',
d_5 decimal(12,2) comment '往前第5天均价',
d_4 decimal(12,2) comment '往前第4天均价',
d_3 decimal(12,2) comment '往前第3天均价',
d_2 decimal(12,2) comment '往前第2天均价',
d_1 decimal(12,2) comment '往前第1天均价',
d_0 decimal(12,2) comment '统计当日均价',
avg_15d decimal(12,2) comment '15天综合平均价',
days_have_data int comment '15天内有价格记录的天数',
create_time timestamp comment '宽表生成时间',
stat_date string comment '统计基准日期，格式yyyy-MM-dd'
)
comment 'DWS层-商品近15日价格透视汇总宽表，每日动态滚动最近15天'
stored as orc
tblproperties ('orc.compress'='SNAPPY');
truncate table dws_15day_summary;
insert overwrite table dws_15day_summary
select
prod_cat,
prod_pcat,
prod_name,
round(avg(case when diff=14 then avg_price end), 2) as d_14,
round(avg(case when diff=13 then avg_price end), 2) as d_13,
round(avg(case when diff=12 then avg_price end), 2) as d_12,
round(avg(case when diff=11 then avg_price end), 2) as d_11,
round(avg(case when diff=10 then avg_price end), 2) as d_10,
round(avg(case when diff=9 then avg_price end), 2) as d_9,
round(avg(case when diff=8 then avg_price end), 2) as d_8,
round(avg(case when diff=7 then avg_price end), 2) as d_7,
round(avg(case when diff=6 then avg_price end), 2) as d_6,
round(avg(case when diff=5 then avg_price end), 2) as d_5,
round(avg(case when diff=4 then avg_price end), 2) as d_4,
round(avg(case when diff=3 then avg_price end), 2) as d_3,
round(avg(case when diff=2 then avg_price end), 2) as d_2,
round(avg(case when diff=1 then avg_price end), 2) as d_1,
round(avg(case when diff=0 then avg_price end), 2) as d_0,
round(avg(case when diff BETWEEN 0 and 14 then avg_price end), 2) as avg_15d,
count(distinct case when diff BETWEEN 0 and 14 then pub_date end) as days_have_data,
current_timestamp as create_time,
-- 基准统计日期：执行当天完整年月日
date_format(current_date(), 'yyyy-MM-dd') as stat_date
from (
select
*,
datediff(current_date(), pub_date) as diff
from dwd_xinfadi_data
where datediff(current_date(), pub_date) between 0 and 14
) tmp
group by prod_cat, prod_pcat, prod_name;
select count(*) as cont from dws_15day_summary;
select * from dws_15day_summary where stat_date='2026-06-27' limit 5;
select d_14,d_13,d_12,d_11,d_10,d_9,d_8,d_7,d_6,d_5,d_4,d_3,d_2,d_1,d_0 from dws_15day_summary limit 5;
select place,count(*) from dwd_xinfadi_data group by place order by place
5.4 dws_overall_product_day （单品日汇总表）
create table if not exists dws_overall_product_day (
pub_date DATE comment '交易日期',
prod_cat string comment '一级品类',
prod_pcat string comment '二级品类',
prod_name string comment '单品名称',
avg_price decimal(10,2) comment '当日单品平均价格',
min_price decimal(10,2) comment '当日单品最低价格',
max_price decimal(10,2) comment '当日单品最高价格'
)
comment 'DWS-每日单品维度汇总宽表(含单品名，供涨跌柱状图使用)'
stored as orc
tblproperties ('orc.compress'='SNAPPY');
insert overwrite table dws_overall_product_day
select
pub_date,
prod_cat,
prod_pcat,
prod_name,
round(avg(avg_price), 2) as avg_price,
round(min(avg_price), 2) as min_price,
round(max(avg_price), 2) as max_price
from dwd_xinfadi_data
group by pub_date, prod_cat, prod_pcat, prod_name;
select * from dws_overall_product_day where pub_date= '2026-06-21' limit 5;
6、ads 应用层：报表表 / 接口表 / 大屏数据
6.1 ads_market_today_index 当日核心数字指标（3 个大屏顶部数字）
-- date_format(日期, 格式模板)
-- concat(字段1,字段2) 是 Hive 字符串拼接函数，把两个字段的值拼成一整条字符串。
-- 1.1 建表语句
create table if not exists ads_market_today_index (
stat_date string comment '统计日期 yyyy-MM-dd',
total_category_cnt int comment '当日一级商品总类别总数',
second_category_cnt int comment '当日二级商品类别总数',
market_avg_price decimal(12,2) comment '全市场当日平均价格指数',
create_time timestamp comment '数据生成时间'
)
comment 'ADS-当日市场核心数字指标（大屏顶部数字卡片）'
stored as orc;
truncate table ads_market_today_index;
-- 插入 SQL（不限制单天，读取全部存在的数据，自动按日期分组）
insert overwrite table ads_market_today_index
select
date_format(t.pub_date, 'yyyy-MM-dd') as stat_date,
count(distinct t.prod_cat) as total_category_cnt,
count(distinct concat(t.prod_cat, t.prod_pcat)) as second_category_cnt,
round(avg(t.avg_price), 2) as market_avg_price,
current_timestamp as create_time
from dws_overall_day t
group by date_format(t.pub_date, 'yyyy-MM-dd');
-- 增量插入 重复传入相同日期的数据，会报错 直接OVERWRITE
insert into table ads_market_today_index
select
'2026-06-22' as stat_date,
count(distinct t.prod_cat) as total_category_cnt,
count(distinct concat(t.prod_cat, t.prod_pcat)) as second_category_cnt,
round(avg(t.avg_price), 2) as market_avg_price,
current_timestamp as create_time
from dws_overall_day t
where date_format(t.pub_date, 'yyyy-MM-dd') = '2026-06-22'
group by date_format(t.pub_date, 'yyyy-MM-dd');
select cont(*) from ads_market_today_index;
select * from ads_market_today_index limit 5
select * from ads_market_today_index where stat_date='2026-06-22';
6.2 ads_market_price_distribution 价格区间饼图
create table if not exists ads_market_price_distribution (
stat_date string comment '统计日期',
price_range string comment '价格区间',
record_count int comment '区间供货记录数',
product_count int comment '区间商品种类数',
ratio decimal(5,2) comment '区间占比%',
create_time timestamp
)
comment 'ADS-每日价格区间分布 饼图数据源'
stored as orc;
insert overwrite table ads_market_price_distribution
select
date_format(pub_date, 'yyyy-MM-dd') as stat_date,
price_range,
record_count,
sum(record_count) over(partition by date_format(pub_date, 'yyyy-MM-dd')) as record_total,
round(record_count / sum(record_count) over(partition by date_format(pub_date, 'yyyy-MM-dd')) * 100, 2) as ratio,
current_timestamp as create_time
from dws_price_range_day;
insert overwrite table ads_market_price_distribution
select
'2026-06-22' as stat_date,
price_range,
record_count,
sum(record_count) over() as record_total,
round(record_count / sum(record_count) over() * 100, 2) as ratio,
current_timestamp as create_time
from dws_price_range_day
where date_format(pub_date, 'yyyy-MM-dd') = '2026-06-22';
select * from ads_market_price_distribution limit 20 ;
6.3 ads_market_day_rise_fall 当日涨幅 / 跌幅柱状图共用表
create table if not exists ads_market_day_rise_fall (
stat_date string comment '统计日期',
prod_cat string comment '一级分类',
prod_pcat string comment '二级分类',
prod_name string comment '商品名称',
yesterday_avg decimal(12,2) comment '前一日均价',
today_avg decimal(12,2) comment '当日均价',
price_diff decimal(12,2) comment '涨跌额',
rise_ratio decimal(5,2) comment '涨跌幅百分比',
flag string comment 'rise涨幅 / fall跌幅',
create_time timestamp
)
comment 'ADS-当日商品涨跌明细，涨幅、跌幅柱状图共用'
stored as orc;
insert overwrite table ads_market_day_rise_fall
select
'2026-06-22' as stat_date,
today.prod_cat,
today.prod_pcat,
today.prod_name,
yesterday.avg_price as yesterday_avg,
today.avg_price as today_avg,
round(today.avg_price - yesterday.avg_price, 2) as price_diff,
case
when yesterday.avg_price = 0 then 0
else round((today.avg_price - yesterday.avg_price) / yesterday.avg_price * 100, 2)
end as rise_ratio,
case
when today.avg_price > yesterday.avg_price then 'rise'
else 'fall'
end as flag,
current_timestamp as create_time
from dws_overall_product_day today
left join dws_overall_product_day yesterday
on today.prod_cat = yesterday.prod_cat
and today.prod_pcat = yesterday.prod_pcat
and today.prod_name = yesterday.prod_name
and yesterday.pub_date = date_sub('2026-06-22', 1)
where today.pub_date = '2026-06-22'
and yesterday.avg_price is not null ;
select * from ads_market_day_rise_fall limit 5;
6.4 ads_market_trend_line 所有折线趋势统一宽表
-- 4.1 建表语句
create table if not exists ads_market_trend_line (
stat_date string comment '统计基准日(15天窗口终点)',
line_name string comment '折线名称',
day_0 decimal(12,2),
day_1 decimal(12,2),
day_2 decimal(12,2),
day_3 decimal(12,2),
day_4 decimal(12,2),
day_5 decimal(12,2),
day_6 decimal(12,2),
day_7 decimal(12,2),
day_8 decimal(12,2),
day_9 decimal(12,2),
day_10 decimal(12,2),
day_11 decimal(12,2),
day_12 decimal(12,2),
day_13 decimal(12,2),
day_14 decimal(12,2),
create_time timestamp
)
comment 'ADS-7日/15日全部折线图时序宽表'
stored as orc;
insert overwrite table ads_market_trend_line
select
'2026-06-27' as stat_date,
prod_cat as line_name,
round(avg(d_0), 2) as day_0,
round(avg(d_1), 2) as day_1,
round(avg(d_2), 2) as day_2,
round(avg(d_3), 2) as day_3,
round(avg(d_4), 2) as day_4,
round(avg(d_5), 2) as day_5,
round(avg(d_6), 2) as day_6,
round(avg(d_7), 2) as day_7,
round(avg(d_8), 2) as day_8,
round(avg(d_9), 2) as day_9,
round(avg(d_10), 2) as day_10,
round(avg(d_11), 2) as day_11,
round(avg(d_12), 2) as day_12,
round(avg(d_13), 2) as day_13,
round(avg(d_14), 2) as day_14,
current_timestamp as create_time
from dws_15day_summary
where stat_date = '2026-06-27'
and prod_cat IN ('蔬菜','肉禽蛋','水果')
group by prod_cat;
select * from ads_market_trend_line limit 3;