Hive SQL 进阶与优化 · 数据开发学习站

本章目标：掌握 lateral view explode、collect_set、get_json_object、日期和空值函数、窗口函数，能独立完成“订单多商品拆行”，并说清 Hive 的常见优化手段。前置知识：Hive 基础、Hive 分区、窗口函数预计用时：110 分钟

离线数仓里，源数据经常是“一行塞了多个商品”或“一行是一个 JSON”。必须先拆行、解析、再聚合。本章用一个订单 JSON 的完整例子，把 Hive 最常用的复杂类型和优化串起来。

1. 场景：一行里藏着多个商品

业务库同步过来的订单常把多个商品塞进一行 JSON。数仓明细要求“一行一个商品”，不做拆行，SUM(price * qty) 就没法按商品维度统计。

Hive 提供了三层工具：get_json_object / from_json 把 JSON 变成字段和数组；lateral view explode 把数组拆成多行；collect_set / collect_list 反方向，把多行合并成数组。

2. 复杂类型与 explode

2.1 三种复杂类型

类型	写法	例子
数组	`ARRAY<STRING>`	`['P3000','P3001']`
映射	`MAP<STRING,INT>`	`{'P3000':2,'P3001':1}`
结构体	`STRUCT<product_id:STRING,qty:INT>`	`{product_id:'P3000', qty:2}`

复杂类型可以嵌套，例如 ARRAY<STRUCT<product_id:STRING,qty:INT,price:DOUBLE>>，正好对应订单里的 items。

2.2 explode：一行拆多行

SELECT order_id, product_id
FROM ods_order_concat
LATERAL VIEW explode(split(product_ids, ',')) t AS product_id;

product_ids = 'P3000,P3001' 会拆成两行。explode 是 UDTF（一行进、多行出），Hive 里不能把它直接写在 SELECT 里再带其他列，必须用 LATERAL VIEW。

2.3 posexplode：带序号

SELECT order_id, pos + 1 AS item_seq, product_id
FROM ods_order_concat
LATERAL VIEW posexplode(split(product_ids, ',')) t AS pos, product_id;

pos 从 0 开始，加 1 就是明细序号 item_seq。拆行后要保留原始顺序时用 posexplode。

2.4 拆行后的行数变化

一行拆成 N 行后，COUNT(*) 从“订单数”变成“商品行数”；如果 price 是订单总价，SUM(price) 会被重复加 N 次。订单级指标要用 COUNT(DISTINCT order_id)，或在拆行前先算好订单金额。

注意：explode 后先确认粒度，再做聚合。

3. collect_set 与 collect_list

3.1 多行合并成数组

SELECT
    customer_id,
    collect_set(category) AS categories,
    size(collect_set(category)) AS category_cnt,
    concat_ws(',', collect_list(product_id)) AS product_list
FROM dwd_order_detail
GROUP BY customer_id;

collect_set：去重，返回数组，顺序不保证。
collect_list：不去重，保留所有值，顺序也不保证。
size：数组长度。
concat_ws：把数组合并成字符串。

需要稳定顺序时先排序：

SELECT customer_id,
       concat_ws(',', sort_array(collect_list(product_id))) AS product_list
FROM dwd_order_detail
GROUP BY customer_id;

典型用途包括用户兴趣标签、订单商品列表、会话页面路径。collect_set 会把数据收进内存，高基数、超长数组可能导致 OOM，只收集必要的列。

4. get_json_object 与 JSON 解析

4.1 取单个字段

SELECT
    get_json_object(line, '$.order_id') AS order_id,
    get_json_object(line, '$.customer_id') AS customer_id,
    get_json_object(line, '$.items[0].product_id') AS first_product
FROM ods_order_json;

$ 表示根节点，.items[0] 表示数组第一个元素。JSON 非法时返回 NULL，不会报错，所以后续要检查空值率。

4.2 解析数组

SELECT
    o.order_id,
    item.product_id,
    item.qty,
    item.price
FROM (
    SELECT
        get_json_object(line, '$.order_id') AS order_id,
        from_json(
            get_json_object(line, '$.items'),
            'array<struct<product_id:string,qty:int,price:double>>'
        ) AS items
    FROM ods_order_json
) o
LATERAL VIEW explode(o.items) t AS item;

from_json 把 JSON 字符串转成 Hive 的数组结构，explode 再拆行。取结构体字段用 item.product_id。

注意：from_json 的 schema 必须和 JSON 完全对应。字段名大小写、类型不匹配都会得到 NULL，不会报错。上线前先用几行数据验证。

5. 日期、空值与窗口函数

5.1 日期函数

需求	Hive 写法
加减天数	`date_add(d, 7)` / `date_sub(d, 1)`
日期差	`datediff(d1, d2)`
格式化为字符串	`date_format(d, 'yyyy-MM')`
当月最后一天	`last_day(d)`
当月第一天	`trunc(d, 'MM')`
字符串转日期	`to_date('2018-06-01')`
时间戳转日期	`from_unixtime(ts, 'yyyy-MM-dd')`

SELECT
    date_add(order_purchase_timestamp, 7) AS plus_7d,
    datediff(order_delivered_customer_date, order_purchase_timestamp) AS delivery_days,
    date_format(order_purchase_timestamp, 'yyyy-MM') AS month
FROM ods_orders
WHERE order_purchase_timestamp IS NOT NULL;

5.2 nvl、coalesce、nullif

SELECT
    nvl(customer_city, '未知') AS city_1,
    coalesce(customer_city, customer_state, '未知') AS region,
    nullif(customer_city, '') AS city_2
FROM ods_customers;

nvl(a, b)：a 为 NULL 返回 b，只接受两个参数。
coalesce(a, b, c, ...)：返回第一个非 NULL 值。
nullif(a, b)：a 等于 b 时返回 NULL，常用于把空字符串转成 NULL。

统计时要注意：COUNT(column) 不统计 NULL，COUNT(*) 统计所有行。SUM 遇到全 NULL 返回 NULL，不是 0，必要时用 COALESCE(SUM(x), 0)。

5.3 窗口函数：环比

WITH monthly AS (
    SELECT date_format(order_date, 'yyyy-MM') AS month,
           SUM(item_amount_with_freight) AS gmv
    FROM dwd_order_detail
    GROUP BY date_format(order_date, 'yyyy-MM')
),
with_prev AS (
    SELECT month, gmv,
           LAG(gmv) OVER (ORDER BY month) AS prev_gmv
    FROM monthly
)
SELECT
    month,
    ROUND(gmv, 2) AS gmv,
    ROUND(prev_gmv, 2) AS prev_gmv,
    ROUND((gmv - prev_gmv) * 100.0 / prev_gmv, 2) AS mom_pct
FROM with_prev
ORDER BY month;

Hive 不允许在同一个 SELECT 里引用刚定义的别名 prev_gmv，所以要再用一层 CTE 或子查询。

分组 TopN 用 ROW_NUMBER() OVER (PARTITION BY dt ORDER BY gmv DESC) 套一层子查询取 rn <= 3。ROW_NUMBER 严格排名，RANK 并列跳号，DENSE_RANK 并列不跳号，区别见窗口函数。

6. Hive UDF 概念

内置函数不够用时，可以写自定义函数。

类型	输入输出	例子
UDF	一行一列进，一个值出	手机号脱敏、身份证校验
UDAF	多行进，一个值出	中位数、自定义聚合
UDTF	一行进，多行出	`explode`、JSON 拆行

6.1 注册 UDF

Java 编写并打 jar 包后：

CREATE FUNCTION mask_phone AS 'com.example.MaskPhone'
USING JAR 'hdfs:///udf/mask-phone.jar';

CREATE TEMPORARY FUNCTION 只在当前会话有效；CREATE FUNCTION 会注册到 Metastore，成为永久函数。

Hive 也可以用 TRANSFORM ... USING 'python3 mask_phone.py' 逐行调用脚本，但启动进程有开销，Spark 里更常用 pandas_udf。

注意：UDF 会绕过 Hive 的向量化执行，简单逻辑优先用内置函数。能用 CASE WHEN、COALESCE、regexp_replace 解决的，不要写 UDF。

7. 完整示例：订单多商品拆行

7.1 建源表和明细表

CREATE TABLE ods_order_json (
    line STRING
);

CREATE TABLE dwd_order_json_detail (
    order_id      STRING,
    item_seq      INT,
    order_date    DATE,
    customer_id   STRING,
    product_id    STRING,
    category      STRING,
    price         DECIMAL(12, 2),
    quantity      INT,
    item_amount   DECIMAL(14, 2)
)
PARTITIONED BY (dt STRING)
STORED AS PARQUET;

7.2 准备两行数据

INSERT INTO ods_order_json VALUES
('{"order_id":"O900001","customer_id":"C100021","order_date":"2018-06-01","items":[{"product_id":"P3000","qty":2,"price":100.0},{"product_id":"P3001","qty":1,"price":50.0}]}'),
('{"order_id":"O900002","customer_id":"C101998","order_date":"2018-06-02","items":[{"product_id":"P3002","qty":3,"price":20.0}]}');

7.3 解析、拆行、关联商品

SET hive.exec.dynamic.partition = true;
SET hive.exec.dynamic.partition.mode = nonstrict;

WITH parsed AS (
    SELECT
        get_json_object(line, '$.order_id') AS order_id,
        get_json_object(line, '$.customer_id') AS customer_id,
        CAST(get_json_object(line, '$.order_date') AS DATE) AS order_date,
        from_json(
            get_json_object(line, '$.items'),
            'array<struct<product_id:string,qty:int,price:double>>'
        ) AS items
    FROM ods_order_json
),
exploded AS (
    SELECT
        p.order_id,
        p.customer_id,
        p.order_date,
        t.pos + 1 AS item_seq,
        t.item AS item
    FROM parsed p
    LATERAL VIEW posexplode(p.items) t AS pos, item
)
INSERT OVERWRITE TABLE dwd_order_json_detail PARTITION (dt)
SELECT
    e.order_id,
    e.item_seq,
    e.order_date,
    e.customer_id,
    e.item.product_id AS product_id,
    COALESCE(d.category, 'unknown') AS category,
    CAST(e.item.price AS DECIMAL(12, 2)) AS price,
    e.item.qty AS quantity,
    CAST(e.item.price * e.item.qty AS DECIMAL(14, 2)) AS item_amount,
    date_format(e.order_date, 'yyyy-MM-dd') AS dt
FROM exploded e
LEFT JOIN ods_products d ON e.item.product_id = d.product_id;

注意：Hive 的 CTE 后面直接接 INSERT OVERWRITE 的写法在部分版本不支持。如果报语法错误，把 INSERT OVERWRITE 放在最前面，把 WITH 放进 SELECT 子查询，或者先建临时表。

7.4 验证结果

SELECT COUNT(*) AS detail_rows,
       COUNT(DISTINCT order_id) AS order_cnt,
       ROUND(SUM(item_amount), 2) AS amount
FROM dwd_order_json_detail;

SHOW PARTITIONS dwd_order_json_detail;

预期 3 行明细、2 个订单、金额 310.00（2×100 + 1×50 + 3×20），以及 dt=2018-06-01、dt=2018-06-02 两个分区。

8. Hive 优化

8.1 分区裁剪和列裁剪

SELECT dt, SUM(item_amount)
FROM dwd_order_detail
WHERE dt = '2018-06-01'
GROUP BY dt;

只扫一个分区、只读 dt 和 item_amount 两列。不要写 SELECT *，不要对分区列做函数运算。

8.2 JOIN 优化

小表广播：

SELECT /*+ MAPJOIN(d) */ f.order_id, d.category
FROM fact_order_item f
JOIN dim_product d ON f.product_id = d.product_id;

小表进内存，避免大表 shuffle。Spark 里叫 broadcast join，参数是 spark.sql.autoBroadcastJoinThreshold。

桶 Map JOIN / SMB JOIN：两表按 JOIN 键分桶，桶数相同或成倍数，且按 JOIN 键排序，可以直接归并。见 Hive 分区。

空值处理：JOIN 键为 NULL 的行会全部落到同一个 reducer，能过滤就加 WHERE f.customer_id IS NOT NULL。数据倾斜可先开 hive.optimize.skewjoin=true，或手动加盐做两阶段聚合：

WITH salted AS (
    SELECT CONCAT(customer_id, '_', CAST(FLOOR(RAND() * 10) AS INT)) AS salt_key, amount
    FROM fact_order
)
SELECT split(salt_key, '_')[0] AS customer_id, SUM(amount) AS amount
FROM salted
GROUP BY split(salt_key, '_')[0];

先按加盐 key 局部聚合，再按真实 key 汇总；复杂场景需要嵌套两层聚合。

8.3 小文件

动态分区、过多 reducer、频繁小批量写入都会产生小文件。Hive 开 hive.merge.mapfiles、hive.merge.mapredfiles 和 hive.merge.size.per.task 做任务结束合并；Spark 用 repartition / coalesce 控制输出文件数，详见 Hive 分区。

8.4 其他常用手段

尽早过滤、用 UNION ALL 代替 UNION、多个 COUNT(DISTINCT) 改写成 GROUP BY 子查询、JOIN 键类型保持一致、开启 hive.map.aggr=true。上线前用 EXPLAIN 确认分区裁剪、JOIN 方式和 shuffle。

9. 常见坑

explode 后直接 COUNT(*)：得到的是商品行数，不是订单数。
JSON schema 写错：from_json 返回 NULL，不报错。
collect_set 内存爆：高基数数组把 executor 撑爆。
窗口函数别名复用：Hive 同一层不能引用刚定义的别名，要套一层子查询。
JOIN 键 NULL 造成倾斜：大量 NULL 落到一个 reducer。
拆行后金额重复加：订单总价在每行都存一份，SUM 会放大。

10. 在线练习

题目	考点
180. 连续出现的数字	窗口函数
184. 部门工资最高的员工	分组取 TopN
185. 部门工资前三高的所有员工	`DENSE_RANK`
1321. 餐馆营业额变化增长	窗口聚合

自测：

能用 explode / posexplode 把一行多商品拆成多行。
能说出 collect_set 和 collect_list 的区别。
能用 get_json_object 取出嵌套 JSON 字段。
能用 ROW_NUMBER 写一个分组去重查询。

11. 面试延伸

面试题：lateral view explode 之后数据量变大，怎么保证指标不重复？

是什么：explode 是 UDTF，一行数组拆成多行，会把原表行数放大。

解决什么问题：拆行前先明确粒度。订单金额是订单级度量，拆行后不能直接 SUM；应先按订单算好金额再拆，订单数用 COUNT(DISTINCT order_id)。

我的实操边界：我在本地验证了 2 个订单拆成 3 行、金额 310；没有处理过超大数组的拆行倾斜。

面试题：UDF、UDAF、UDTF 有什么区别？

是什么：UDF 一行一值，UDAF 多行一值，UDTF 一行多值。

解决什么问题：UDF 做字段级转换，UDAF 做自定义聚合，UDTF 拆数组或 JSON。UDF 通常由 Java 编写并注册 jar，Python 可以用 TRANSFORM。

我的实操边界：我了解三类函数的输入输出和注册方式，项目里主要用内置函数和 explode，没有发布过 Java UDF。

面试题：Hive 数据倾斜怎么优化？

是什么：数据倾斜是某些 key 的数据量远大于其他 key，导致少数 task 拖慢整个作业。

解决什么问题：先定位热点 key，再按场景选择：过滤 NULL、广播小表、热点 key 单独处理、加盐两阶段聚合、开启 skewjoin。不同原因对应不同手段，不能只会加随机数。

我的实操边界：我在本地用加盐 SQL 演示过两阶段聚合，理解 Map JOIN 和 SMB JOIN 的条件；没有在生产集群上通过 Spark UI 定位过真实的倾斜任务。

12. 本章验收

能用 LATERAL VIEW explode 和 posexplode 把数组拆成多行
能说出 explode 后 COUNT(*) 和 COUNT(DISTINCT order_id) 的区别
能用 collect_set 统计用户品类集合，并说明内存风险
能用 get_json_object、from_json 解析 JSON 和数组
能写日期函数、coalesce、窗口环比和分组 TopN 的 SQL
能独立完成订单 JSON 拆行并写入分区表，核对行数和金额
能说出分区裁剪、Map JOIN、加盐、小文件治理的适用场景