当前位置：首页 > news >正文

6.1.2.2 大数据方法论与实践指南-离线任务SQL 任务开发规范

news 2025/10/29 8:19:21

6.1.2.2 离线 SQL 任务开发规范

大数据离线 SQL 任务（如 Hive SQL、Spark SQL）是数据仓库建设和离线数据分析的核心载体，其开发质量直接直接直接规范直接直接影响任务效率、数据质量和可维护性。以下从文件组织、命名规范、SQL 编写、性能优化、数据质量、上线控流程六个维度，提供详细的离线 SQL 任务开发规范。

一、文件组织规范

采用 “数据分层 + 业务域” 二维结构，与数据仓库架构严格对齐，确保脚本可追溯、易管理。

Plain Text
sql/
├── ods/               # 原始数据层（Extract）
│   ├── user/          # 业务域：用户
│   │   ├── ods_user_login_di.sql       # 每日增量：用户登录日志
│   │   └── ods_user_register_df.sql    # 每日全量：用户注册信息
│   └── order/         # 业务域：订单
│       └── ods_order_info_di.sql       # 每日增量：订单信息
├── dwd/               # 明细数据层（Transform-清洗）
│   ├── user/
│   │   └── dwd_user_login_detail_di.sql # 用户登录明细
│   └── order/
│       └── dwd_order_detail_di.sql      # 订单明细
├── dws/               # 汇总数据层（Transform-聚合）
│   ├── user/
│   │   └── dws_user_login_agg_dd.sql    # 用户登录日汇总
│   └── order/
│       └── dws_order_agg_dd.sql         # 订单日汇总
└── ads/               # 应用数据层（Load）
└── ads_sales_summary_dd.sql         # 销售汇总报表

一级目录：按数据分层（ODS/DWD/DWS/ADS）划分，对应数据加工阶段；

二级目录：按业务域（user/order/payment 等）划分，隔离不同业务数据；

文件粒度：单个 SQL 文件只负责一张表的生成，禁止一个文件生成多表。

任务粒度：单个任务只执行一个 SQL 文件。

二、命名规范

统一命名风格，提升可读性和一致性。

SQL 文件命名（任务命名除了后缀名其它一致）

与目标表名保持一致，格式：{表名}.sql，如dwd_order_detail_di.sql。

变量命名

日期变量：biz_date（业务日期，如2025-09-06）、pre_date（前一天）、start_date（开始日期）；

阈值变量：max_null_ratio（最大空值率）、valid_threshold（有效数据阈值）。

三、SQL 编写规范

头部注释

每个 SQL 文件必须包含头部注释，说明核心信息：

SQL
-- 目标表：dwd.order_detail_di
-- 功能：清洗订单原始数据，生成订单明细（每日增量）
-- 输入表：ods.order_info_di（当日增量）、dim.order_status（订单状态维表）
-- 输出字段：order_id（订单ID）、user_id（用户ID）、order_amount（订单金额）、pay_status（支付状态）等
-- 加工逻辑：1. 过滤无效订单（order_id为空）；2. 关联维表转换状态码为状态名；3. 脱敏手机号
-- 调度周期：每日凌晨3点
-- 创建人：zhang-san
-- 创建时间：2025-09-01
-- 变更记录：2025-09-05 李四新增优惠金额字段
-- request_id: 需求id列表

分区规范

必须指定分区：所有读写操作强制过滤分区字段（如dt = '${biz_date}'），禁止全表扫描；

SQL
-- 推荐：明确分区
INSERT OVERWRITE TABLE dwd.order_detail_di PARTITION (dt = '${biz_date}')
SELECT ...
FROM ods.order_info_di WHERE dt = '${biz_date}';
-- 禁止：无分区过滤（全表扫描）
INSERT OVERWRITE TABLE dwd.order_detail_di SELECT ... FROM ods.order_info_di;

分区字段统一：时间分区用dt（天），格式yyyy-MM-dd；小时级分区加hr（如dt='2025-09-06' AND hr='08'）；

禁止跨分区覆盖：一次任务只处理单个分区（如dt='${biz_date}'），避免误删历史数据。

语法规范

** 禁止 SELECT ***：只查询需要的字段，减少数据传输；

SQL
-- 推荐
SELECT order_id, user_id, create_time FROM ods.order_info_di;
-- 禁止
SELECT * FROM ods.order_info_di;

使用显式字段关联：JOIN时明确关联条件，禁止CROSS JOIN（笛卡尔积）；

SQL
-- 推荐
SELECT a.order_id, b.user_name
FROM dwd.order_detail_di a
LEFT JOIN dim.user_info b
ON a.user_id = b.user_id
-- 显式关联字段
WHERE a.dt = '${biz_date}';
-- 禁止：无关联条件（笛卡尔积）
SELECT a.order_id, b.user_name FROM dwd.order_detail_di a, dim.user_info b;

合理使用 CTE（公用表表达式）：复杂逻辑拆分为 CTE，提高可读性；

SQL
WITH
valid_orders AS (
-- 步骤1：过滤有效订单
SELECT order_id, user_id, order_amount
FROM ods.order_info_di
WHERE dt = '${biz_date}'
AND order_id IS NOT NULL
-- 主键非空
AND order_amount > 0
-- 金额合理),
order_with_status AS (
-- 步骤2：关联状态维表
SELECT a.*, b.status_name
FROM valid_orders a
LEFT JOIN dim.order_status b
ON a.status_code = b.code
)
-- 最终插入目标表
INSERT OVERWRITE TABLE dwd.order_detail_di PARTITION (dt = '${biz_date}')SELECT order_id, user_id, order_amount, status_name FROM order_with_status;

避免隐式类型转换：如字符串与数字比较（order_id = '123'而非order_id = 123）；

聚合函数必带过滤：GROUP BY前先过滤无效数据，减少计算量；

注释清晰：关键逻辑（如复杂判断、业务规则）需加行注释。

写入规范

统一使用 INSERT OVERWRITE：确保任务可重跑（覆盖当天分区，不影响历史）；

一个表只有一个任务更新，确保更新逻辑统一在一处；

一个任务只更新一个表；

目标表字段显式列出：避免因表结构变更导致字段错位；

SQL
-- 推荐：显式字段
INSERT OVERWRITE TABLE dwd.order_detail_di PARTITION (dt = '${biz_date}')(order_id, user_id, order_amount, create_time)SELECT
order_id,
user_id,
order_amount,
create_time
FROM ods.order_info_di
WHERE dt = '${biz_date}';
-- 禁止：依赖字段顺序（表结构变更后易出错）
INSERT OVERWRITE TABLE dwd.order_detail_di PARTITION (dt = '${biz_date}')SELECT order_id, user_id, order_amount, create_time FROM ods.order_info_di;

四、性能优化规范

减少数据扫描

列裁剪：只查询必要字段（避免SELECT *）；

谓词下推：过滤条件尽可能前置（如WHERE条件会被 Hive/Spark 下推到存储层）；

分区过滤优先：WHERE子句中分区字段（dt）放在最前面，加速过滤。

避免数据倾斜

识别倾斜字段：通过GROUP BY字段的COUNT分布判断（如某user_id占比超 50%）；

倾斜处理方案：

加盐法：对热点 Key 添加随机前缀，分散到多个分区；

SQL
-- 对热点user_id加盐（假设user_id=0是热点）
WITH salted_orders AS (SELECT
CASE WHEN user_id = '0' THEN CONCAT(user_id, '_', rand()%10) ELSE user_id END AS salted_user_id,
order_amount
FROM dwd.order_detail_di
WHERE dt = '${biz_date}'),
partial_agg AS (
-- 第一次聚合（分散计算）
SELECT salted_user_id, SUM(order_amount) AS total
FROM salted_orders
GROUP BY salted_user_id
)
-- 第二次聚合（合并结果）
SELECT
CASE WHEN salted_user_id LIKE '0_%' THEN '0' ELSE salted_user_id END AS user_id,SUM(total) AS total_amount
FROM partial_agg
GROUP BY CASE WHEN salted_user_id LIKE '0_%' THEN '0' ELSE salted_user_id END;

广播小表：小表（<1GB）通过/*+ BROADCAST(b) */提示广播，避免 Shuffle；

SQL
-- 广播用户维表（小表）
SELECT /*+ BROADCAST(b) */
a.order_id, b.user_name
FROM dwd.order_detail_di a
LEFT JOIN dim.user_info b
ON a.user_id = b.user_id
WHERE a.dt = '${biz_date}';

小文件处理

写入时控制文件数量：通过distribute by均匀分配数据，避免过多小文件；

SQL
-- Hive：按user_id哈希分区，控制文件数=100
INSERT OVERWRITE TABLE dws.order_agg_dd PARTITION (dt = '${biz_date}')SELECT user_id, SUM(order_amount) AS total
FROM dwd.order_detail_di
WHERE dt = '${biz_date}'GROUP BY user_id
DISTRIBUTE BY user_id;
-- 按用户ID哈希，数据均匀分布

Spark SQL 配置：设置spark.sql.files.maxRecordsPerFile=1000000（每文件约 100 万条）；

定期合并历史小文件：通过调度任务执行ALTER TABLE ... CONCATENATE（Hive）。

索引与存储优化

存储格式：非 ODS 层表强制使用 Parquet/ORC（列存、压缩比高），禁止 TextFile；

压缩算法：默认 Snappy（平衡速度和压缩比），冷数据（30 天以上）可用 GZIP；

分区粒度：单分区大小控制在 1GB～10GB，避免过多小分区（如每小时分区但数据量极少）。

JOIN 优化

小表驱动大表：将小表放在 JOIN 右侧，或使用 MAP JOIN（Hive）或 BROADCAST JOIN（Spark SQL）：

SQL
-- Hive: 启用 Map Join 提示
SELECT /*+ MAPJOIN(small_table) */ a.user_id, b.user_name
FROM large_table a JOIN small_table b ON a.user_id = b.user_id;

-- Spark SQL: 使用广播变量
SET spark.sql.autoBroadcastJoinThreshold=10485760; -- 10MB
SELECT a.user_id, b.user_name
FROM large_table a JOIN small_table b ON a.user_id = b.user_id;

避免大表 JOIN 大表：如必须执行，考虑分治策略（如先聚合再 JOIN）。

数据倾斜处理

热点键处理：对倾斜键（如 user_id=NULL）添加随机前缀：

SQL
-- 示例：处理订单金额统计中的倾斜
SELECT
CASE
WHEN user_id IS NULL THEN CONCAT('null_', FLOOR(RAND() * 10)) -- 随机分桶
ELSE user_id
END AS user_id_key,
SUM(order_amount) AS total_amount
FROM dw_order_detail
GROUP BY
CASE
WHEN user_id IS NULL THEN CONCAT('null_', FLOOR(RAND() * 10))
ELSE user_id
END;

使用 DISTRIBUTE BY：在 Spark SQL 中手动指定分区键：

SQL
SET spark.sql.shuffle.partitions=200; -- 调整分区数
SELECT user_id, SUM(order_amount)
FROM dw_order_detail
DISTRIBUTE BY user_id -- 手动控制数据分布
GROUP BY user_id;

聚合优化

先过滤再聚合：减少聚合数据量：

SQL
-- 正确：先过滤再聚合
SELECT user_id, COUNT(*) AS order_count
FROM dw_order_detail
WHERE dt = '20240101' AND status = 'completed'
GROUP BY user_id;

-- 错误：先聚合再过滤
SELECT user_id, COUNT(*) AS order_count
FROM dw_order_detail
GROUP BY user_id
HAVING dt = '20240101' AND status = 'completed'; -- 无效，HAVING 不能过滤原始字段

避免使用 DISTINCT

优先使用 GROUP BY 替代 DISTINCT，减少计算开销：

SQL
-- 正确：使用 GROUP BY
SELECT user_id FROM dw_order_detail GROUP BY user_id;

-- 错误：使用 DISTINCT
SELECT DISTINCT user_id FROM dw_order_detail;

五、数据质量规范

数据校验

每个任务必须包含数据校验逻辑，校验不通过则任务失败并告警：（调度系统提供数据校验模板）

数据清洗规则

去重：基于唯一键（如order_id）去重，保留最新记录；

SQL
-- 保留最新订单记录（按create_time排序）
INSERT OVERWRITE TABLE dwd.order_detail_di PARTITION (dt = '${biz_date}')SELECT order_id, user_id, order_amount, create_time
FROM (SELECT
*,
row_number() OVER (PARTITION BY order_id ORDER BY create_time DESC) AS rn
FROM ods.order_info_di
WHERE dt = '${biz_date}') t
WHERE rn = 1;