本文目标：建立 SQL 核心能力与大数据开发场景之间的完整认知链路。

一、重新理解 SQL：它不只是查询语言

大多数人把 SQL 理解为"取数工具"，这是认知的第一个坑。SQL 本质上是声明式的关系代数表达，它描述的是"要什么"而非"怎么做"，底层执行计划由优化器决定。理解这一点，是写出高质量 SQL 的前提。

SQL 的核心能力域：

• 集合运算：JOIN、UNION、INTERSECT、EXCEPT，本质是关系代数
• 聚合与窗口：GROUP BY 是分组折叠，Window Function 是分组不折叠
• 子查询与 CTE：逻辑分层，控制执行顺序与可读性
• 条件与过滤：WHERE vs HAVING 的执行时机差异决定性能边界
• 排序与分页：ORDER BY + LIMIT 的下推优化是大数据场景的核心考点

在大数据引擎（Hive、Spark SQL、Flink SQL、Presto/Trino）中，SQL 语义保持一致，但执行模型截然不同。理解这个分层，才能在不同引擎间迁移时不翻车。

二、窗口函数

窗口函数是 SQL 最被低估的特性，也是数据开发中频率最高的"降复杂度"工具。

2.1 核心语法结构

function_name(...) OVER (
    PARTITION BY col1, col2   -- 分组，不折叠行
    ORDER BY col3 DESC        -- 组内排序
    ROWS BETWEEN 2 PRECEDING AND CURRENT ROW  -- 滑动窗口
)

PARTITION BY 决定计算范围，ORDER BY 决定计算顺序，ROWS/RANGE 决定窗口帧。三者缺一不可，混淆任何一个都会得到错误结果。

2.2 最高频场景：去重保留最新一条

业务表中同一 ID 存在多条更新记录，需要取最新状态：

-- 错误做法：GROUP BY + MAX 无法同时取其他字段
-- 正确做法：ROW_NUMBER 窗口函数
SELECT *
FROM (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY updated_at DESC) AS rn
    FROM user_profile
) t
WHERE rn = 1;

2.3 累计值与环比计算(*****)

-- 累计 GMV（running total）
SELECT
    dt,
    gmv,
    SUM(gmv) OVER (ORDER BY dt ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS cumulative_gmv,
    -- 环比增长率
    LAG(gmv, 1) OVER (ORDER BY dt) AS prev_gmv,
    ROUND((gmv - LAG(gmv, 1) OVER (ORDER BY dt)) / LAG(gmv, 1) OVER (ORDER BY dt) * 100, 2) AS mom_rate
FROM dws_gmv_daily;

LAG/LEAD 替代自连接，在 Spark SQL 中能减少一次 Shuffle，性能差距在亿级数据下可达 3-5 倍。

2.4 分组 TopN：经典的大数据场景

-- 每个品类销售额 Top3 的商品
SELECT category, item_id, sales
FROM (
    SELECT
        category,
        item_id,
        sales,
        DENSE_RANK() OVER (PARTITION BY category ORDER BY sales DESC) AS dr
    FROM dws_item_sales
) t
WHERE dr <= 3;

RANK、DENSE_RANK、ROW_NUMBER 的区别：遇到并列时的行为不同。TopN 业务一般用 DENSE_RANK（并列不跳号），去重用 ROW_NUMBER（强制唯一）。

三、JOIN 的本质与大数据场景下的性能陷阱

3.1 JOIN 的执行模型

在分布式引擎中，JOIN 的核心代价是 Shuffle（数据重分布）。理解 JOIN 的两种实现方式：

• Hash Join（Shuffle Hash Join）：将两张表按 JOIN Key 哈希分桶，再逐桶做内存 Hash 表匹配。绝大多数大数据引擎的默认实现。
• Broadcast Join（Map Join）：将小表广播到每个 Executor，大表直接本地匹配，零 Shuffle。

-- Hive 中显式触发 Broadcast Join
SET hive.auto.convert.join=true;
SET hive.mapjoin.smalltable.filesize=25000000; -- 25MB 以内自动广播

-- Spark SQL 中显式 Hint
SELECT /*+ BROADCAST(small_table) */ *
FROM big_table b
JOIN small_table s ON b.key = s.key;

黄金法则：小表（<= 数百 MB）必须走 Broadcast Join，否则是性能 Bug。

3.2 数据倾斜：JOIN 的头号杀手

倾斜的本质：某个 Key 的数据量远超其他 Key，导致部分 Task 处理时间是其他 Task 的几十甚至几百倍。

诊断方式：

-- 先分析 JOIN Key 的分布
SELECT join_key, COUNT(*) AS cnt
FROM big_table
GROUP BY join_key
ORDER BY cnt DESC
LIMIT 100;

解决方案 1：加盐打散（Salting）

-- 将热点 Key 的大表数据打散到 N 份
SELECT b.key, b.val, s.val
FROM (
    SELECT key, val, CONCAT(key, '_', CAST(FLOOR(RAND() * 10) AS STRING)) AS salted_key
    FROM big_table
) b
JOIN (
    -- 小表扩容 N 份
    SELECT CONCAT(key, '_', seq) AS salted_key, val
    FROM small_table
    LATERAL VIEW POSEXPLODE(ARRAY(0,1,2,3,4,5,6,7,8,9)) t AS seq, dummy
) s
ON b.salted_key = s.salted_key;

解决方案 2：NULL 值单独处理

NULL Key 在 JOIN 中全部进入同一 Task，是最隐蔽的倾斜来源：

-- NULL Key 单独走 UNION，非 NULL 正常 JOIN
SELECT a.*, b.val FROM a JOIN b ON a.key = b.key WHERE a.key IS NOT NULL
UNION ALL
SELECT a.*, NULL AS val FROM a WHERE a.key IS NULL;

3.3 多表 JOIN 的顺序优化

Spark/Hive 的查询优化器会尝试重排 JOIN 顺序，但不总是最优。手工规则：

1. 先过滤，后 JOIN（减少参与 JOIN 的数据量）
2. 小表靠右（Hash Join 中右表建 Hash 表）
3. 分区裁剪优先（确保 WHERE 条件能下推到分区过滤）

四、数仓分层：SQL 如何组织复杂的数据流

数仓分层本质上是用 SQL 描述的 ETL DAG，每一层解决一类问题：

ODS（贴源层）→ DWD（明细层）→ DWS（汇总层）→ ADS（应用层）

层次	核心 SQL 操作	目标
ODS→DWD	去重、清洗、标准化类型、解析 JSON	数据质量
DWD→DWS	GROUP BY 聚合、多维 ROLLUP、窗口累计	计算复用
DWS→ADS	宽表 JOIN、业务逻辑过滤、指标计算	业务对齐

4.1 DWD 层的核心 SQL Pattern

场景：处理 Kafka 消费的 JSON 日志，写入明细宽表

INSERT OVERWRITE TABLE dwd_user_event PARTITION (dt='2024-01-15')
SELECT
    get_json_object(raw_log, '$.user_id')                           AS user_id,
    get_json_object(raw_log, '$.event_type')                        AS event_type,
    CAST(get_json_object(raw_log, '$.ts') AS BIGINT)                AS event_ts,
    FROM_UNIXTIME(CAST(get_json_object(raw_log, '$.ts') AS BIGINT)) AS event_time,
    -- 枚举值标准化
    CASE get_json_object(raw_log, '$.platform')
        WHEN 'ios'     THEN 'iOS'
        WHEN 'android' THEN 'Android'
        ELSE 'Unknown'
    END AS platform,
    -- 空值处理：业务逻辑决定 NULL 还是默认值
    COALESCE(get_json_object(raw_log, '$.item_id'), '-1')           AS item_id
FROM ods_raw_log
WHERE dt = '2024-01-15'
  AND get_json_object(raw_log, '$.user_id') IS NOT NULL  -- 过滤无效数据
  AND LENGTH(get_json_object(raw_log, '$.user_id')) > 0;

4.2 DWS 层的核心 SQL Pattern

场景：计算用户当日多维行为汇总

INSERT OVERWRITE TABLE dws_user_behavior_1d PARTITION (dt='2024-01-15')
SELECT
    user_id,
    platform,
    -- 行为计数
    COUNT(1)                                                         AS pv,
    COUNT(DISTINCT CASE WHEN event_type = 'click' THEN item_id END) AS click_item_uv,
    SUM(CASE WHEN event_type = 'purchase' THEN amount ELSE 0 END)   AS purchase_amount,
    -- 行为时间特征
    MIN(event_ts)                                                    AS first_active_ts,
    MAX(event_ts)                                                    AS last_active_ts,
    MAX(event_ts) - MIN(event_ts)                                    AS active_duration_sec
FROM dwd_user_event
WHERE dt = '2024-01-15'
GROUP BY user_id, platform;

注意 COUNT(DISTINCT ...) 在大数据引擎中极耗资源，在超大规模数据下考虑用 approx_count_distinct（HyperLogLog）替代。

五、复杂业务指标的 SQL 拆解方法论

工程实践中，需求往往用业务语言描述，翻译成 SQL 的能力决定了数据开发的上限。

5.1 留存率计算（用户行为分析的经典模型）

业务定义：首日注册用户在第 N 日仍有活跃的比例。

-- Step 1: 获取首次注册日期（用户粒度）
WITH user_first_day AS (
    SELECT user_id, MIN(dt) AS reg_date
    FROM dwd_user_event
    GROUP BY user_id
),
-- Step 2: 关联后续活跃日期，计算间隔天数
retention_raw AS (
    SELECT
        u.reg_date,
        DATEDIFF(e.dt, u.reg_date) AS day_diff,
        COUNT(DISTINCT e.user_id) AS retained_users
    FROM user_first_day u
    JOIN dwd_user_event e ON u.user_id = e.user_id
    WHERE u.reg_date >= '2024-01-01'
      AND DATEDIFF(e.dt, u.reg_date) BETWEEN 0 AND 30
    GROUP BY u.reg_date, DATEDIFF(e.dt, u.reg_date)
),
-- Step 3: 获取各批次注册人数（分母）
reg_cohort AS (
    SELECT reg_date, COUNT(DISTINCT user_id) AS reg_users
    FROM user_first_day
    WHERE reg_date >= '2024-01-01'
    GROUP BY reg_date
)
-- Step 4: 计算留存率
SELECT
    r.reg_date,
    r.day_diff,
    c.reg_users,
    r.retained_users,
    ROUND(r.retained_users / c.reg_users * 100, 2) AS retention_rate
FROM retention_raw r
JOIN reg_cohort c ON r.reg_date = c.reg_date
ORDER BY r.reg_date, r.day_diff;

5.2 漏斗分析（转化路径）

业务定义：用户从曝光 → 点击 → 加购 → 下单的各步转化率。

WITH funnel AS (
    SELECT
        user_id,
        -- 标记各步骤是否到达（当日）
        MAX(CASE WHEN event_type = 'expose'   THEN 1 ELSE 0 END) AS step1_expose,
        MAX(CASE WHEN event_type = 'click'    THEN 1 ELSE 0 END) AS step2_click,
        MAX(CASE WHEN event_type = 'add_cart' THEN 1 ELSE 0 END) AS step3_cart,
        MAX(CASE WHEN event_type = 'purchase' THEN 1 ELSE 0 END) AS step4_purchase
    FROM dwd_user_event
    WHERE dt = '2024-01-15'
    GROUP BY user_id
)
SELECT
    SUM(step1_expose)                                                  AS expose_users,
    SUM(step2_click)                                                   AS click_users,
    SUM(step3_cart)                                                    AS cart_users,
    SUM(step4_purchase)                                                AS purchase_users,
    ROUND(SUM(step2_click)    / SUM(step1_expose)   * 100, 2)         AS expose_to_click,
    ROUND(SUM(step3_cart)     / SUM(step2_click)    * 100, 2)         AS click_to_cart,
    ROUND(SUM(step4_purchase) / SUM(step3_cart)     * 100, 2)         AS cart_to_purchase
FROM funnel;

核心技巧：用 MAX(CASE WHEN ...) 将多行事件"旋转"成用户维度的宽表（行转列），然后在宽表上做聚合运算，避免多次扫描。

六、Flink SQL：流处理场景下的 SQL 思维转变

Flink SQL 把 SQL 能力延伸到实时流，但需要理解与批处理 SQL 的关键差异。

6.1 时间语义：流处理的核心概念

-- 定义 Kafka Source，声明事件时间 + Watermark
CREATE TABLE kafka_user_event (
    user_id     BIGINT,
    event_type  STRING,
    event_ts    TIMESTAMP(3),
    WATERMARK FOR event_ts AS event_ts - INTERVAL '5' SECOND  -- 允许 5 秒乱序
) WITH (
    'connector' = 'kafka',
    'topic' = 'user_event',
    'format' = 'json'
);

Watermark 的本质：系统用来判断"某个时间窗口的数据已经到齐"的水位线。太紧（0 秒）会丢数据，太宽（60 秒）会增加延迟，需要根据业务容忍度权衡。

6.2 窗口聚合：Tumble vs Hop vs Session

-- 滚动窗口（无重叠）：每 5 分钟统计一次 PV
SELECT
    TUMBLE_START(event_ts, INTERVAL '5' MINUTE) AS window_start,
    event_type,
    COUNT(1) AS pv
FROM kafka_user_event
GROUP BY
    TUMBLE(event_ts, INTERVAL '5' MINUTE),
    event_type;

-- 滑动窗口（有重叠）：过去 10 分钟内、每 1 分钟滑动一次
SELECT
    HOP_START(event_ts, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE) AS window_start,
    COUNT(DISTINCT user_id) AS uv
FROM kafka_user_event
GROUP BY
    HOP(event_ts, INTERVAL '1' MINUTE, INTERVAL '10' MINUTE);

6.3 双流 JOIN：实时维表关联

-- 实时事实表 JOIN 维表（MySQL CDC 作为维表）
SELECT
    e.user_id,
    e.event_type,
    u.user_name,
    u.user_level
FROM kafka_user_event e
LEFT JOIN mysql_user_dim FOR SYSTEM_TIME AS OF e.event_ts AS u
ON e.user_id = u.user_id;

FOR SYSTEM_TIME AS OF 是 Flink SQL 的时态 JOIN（Temporal Join），保证关联到事件发生时刻的维表快照，而非当前最新版本，这对计算历史数据的一致性至关重要。

七、SQL 调优：性能问题的系统化排查框架

性能问题 80% 来自以下几类，按排查优先级排序：

7.1 分区裁剪失效

-- 错误：对分区字段做了函数运算，导致全表扫描
WHERE DATE_FORMAT(dt, 'yyyy-MM') = '2024-01'

-- 正确：让分区字段保持原始形式
WHERE dt BETWEEN '2024-01-01' AND '2024-01-31'

排查方法：看执行计划中 Input 的分区数量，如果是全量分区，则分区裁剪失效。

7.2 COUNT DISTINCT 的替代方案

-- 原始写法：全局去重，单 Reducer 瓶颈
SELECT COUNT(DISTINCT user_id) FROM dwd_user_event WHERE dt = '2024-01-15';

-- 优化 1：两阶段聚合
SELECT COUNT(1)
FROM (SELECT DISTINCT user_id FROM dwd_user_event WHERE dt = '2024-01-15') t;

-- 优化 2：近似计算（误差 <2%，速度提升 10x+）
SELECT approx_count_distinct(user_id) FROM dwd_user_event WHERE dt = '2024-01-15';

7.3 小文件问题

小文件是大数据平台的慢性病：Hive/Spark 每个文件对应一个 Task，大量小文件导致调度开销远超实际计算开销。

-- Hive 写入时合并小文件
SET hive.merge.mapfiles=true;
SET hive.merge.mapredfiles=true;
SET hive.merge.size.per.task=256000000;  -- 合并目标大小 256MB

-- Spark 写入时控制文件数量
-- 按照目标文件大小反推分区数：总数据量 / 目标文件大小
df.repartition(200).write.partitionBy("dt").parquet(path)

7.4 读懂执行计划

-- Spark SQL 查看执行计划
EXPLAIN EXTENDED
SELECT ...;

-- 关注以下关键词：
-- FileScan: 分区数量 → 判断分区裁剪是否生效
-- BroadcastHashJoin vs SortMergeJoin → 判断 Join 策略
-- Exchange hashpartitioning → Shuffle 点，数量越少越好
-- HashAggregate → 聚合是否在 Executor 本地完成（避免全局单点）

八、数据质量：被忽视的 SQL 工程能力

数据开发不只是写 ETL，数据质量保障是核心工程能力。

8.1 常见数据质量 SQL 检查

-- 1. 主键唯一性校验
SELECT user_id, COUNT(*) AS cnt
FROM dwd_user_info
WHERE dt = '2024-01-15'
GROUP BY user_id
HAVING cnt > 1;

-- 2. 关键字段空值率
SELECT
    COUNT(1) AS total,
    SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) AS user_id_null_cnt,
    ROUND(SUM(CASE WHEN user_id IS NULL THEN 1 ELSE 0 END) / COUNT(1) * 100, 4) AS null_rate
FROM dwd_user_event
WHERE dt = '2024-01-15';

-- 3. 数据量环比波动检测（突增突降告警）
SELECT
    dt,
    row_cnt,
    LAG(row_cnt, 1) OVER (ORDER BY dt) AS prev_row_cnt,
    ROUND((row_cnt - LAG(row_cnt, 1) OVER (ORDER BY dt))
          / LAG(row_cnt, 1) OVER (ORDER BY dt) * 100, 2) AS change_rate
FROM (
    SELECT dt, COUNT(1) AS row_cnt
    FROM dwd_user_event
    WHERE dt >= DATE_SUB(CURRENT_DATE, 7)
    GROUP BY dt
) t
ORDER BY dt;

数据量环比波动超过 ±30% 通常需要告警介入，这类 SQL 应封装为每日质量检测任务。