Hive SQL 写递归计算一直是数仓工程师绕不过去的一道坎。业务方一句“我要看全链路组织架构”或者“把这个 BOM 的末级物料全部展开”落到 Hive SQL 上就是一场噩梦。MySQL 里一条WITH RECURSIVE能搞定的事在 Hive 里就是另一套逻辑。我之前用“写死 20 层自连接”的方式凑合过一版姑且叫 1.0后来业务层级超过 20 层SQL 直接废掉才认真做了这套 2.0 方案。这套 2.0 的核心思路并不复杂临时表 循环迭代每一轮只扫增量数据再用一张总表防环路、去重复。实测下来千万级节点、几十层深度的场景都能稳定跑完更重要的是不会因为某个环导致任务跑到天荒地老。这篇文章就完整记录一下这套方案的设计思路、完整代码、性能调优和排坑实录给还在 Hive 里硬刚递归的朋友们一个可以直接抄的作业。1. 为什么需要一套 2.0 方案Hive 本质上是为分布式批处理设计的标准的 SQL 递归查询能力一直不是它的强项。低版本 Hive 根本不支持WITH RECURSIVE新版本虽然借助 Calcite 加了这个语法但生产环境里依赖它跑大规模递归任务的人依然很少后面我会具体讲原因。所以大部分数仓团队实际上都在用“土办法”实现递归。1.1 1.0 方案写死层数的自连接1.0 版本的思路很直白既然不知道最深多少层那就假设一个最大值比如 20 层然后用LEFT JOIN一层层往上拼最后用UNION ALL合并所有层的结果。这种方案应付有限的、明确已知层级的需求确实没问题。我最早处理一个“三二级部门展开”的需求时就是这么干的。但问题也很明显深度是写死的。一旦需求变成“动态层级”SQL 就废了。SQL 冗长得可怕。20 层JOIN的 SQL 要写几百行维护起来头大。性能不稳定。每一层JOIN都会对上一步的全量结果再扫一遍越到后面越慢。1.2 2.0 方案要解决的核心问题2.0 在设计之初定了几个明确目标支持任意深度最多只受集群资源和参数限制。每一轮迭代只处理新发现的增量数据而不是全量重算。防环路。不能因为 A→B→C→A 这种环导致死循环。每一轮结果都带有层级标记方便业务侧按“第几层”做过滤和分析。这套方案从设计、编码到调优大概花了一周落地之后稳定跑了很久也经历了多次数据量和深度翻倍的考验。下面把方案完整拆开讲。2. 递归计算的主流实现路线对比在做 2.0 之前我把当时业界常见的几种 Hive 递归方案都捋了一遍这里直接给结论方便你做技术选型。2.1 写死层数的自连接方案就是 1.0 的做法适用于层级明确且较浅的场景。优点是实现简单、不依赖任何脚本缺点是灵活性差、性能不可控。-- 示例展开 3 层部门 SELECT t1.dep_id AS root_dep_id, t3.dep_id AS child_dep_id, CASE WHEN t2.dep_id IS NULL THEN 1 WHEN t3.dep_id IS NULL THEN 2 ELSE 3 END AS depth FROM ods_dep_rel t1 LEFT JOIN ods_dep_rel t2 ON t1.dep_id t2.parent_dep_id LEFT JOIN ods_dep_rel t3 ON t2.dep_id t3.parent_dep_id WHERE t1.parent_dep_id IS NULL;层数少的时候还能忍层数多起来这段 SQL 会膨胀到你不想维护。2.2 原生 WITH RECURSIVE 方案从 Hive 4.0 开始社区通过 Calcite 引入了原生递归 CTE 支持。理论上你可以写这样的语句WITH RECURSIVE dep_tree AS ( SELECT dep_id, parent_dep_id, dep_name, 0 AS loop_id FROM ods_dep_rel WHERE parent_dep_id IS NULL UNION ALL SELECT c.dep_id, c.parent_dep_id, c.dep_name, t.loop_id 1 FROM dep_tree t JOIN ods_dep_rel c ON t.dep_id c.parent_dep_id ) SELECT * FROM dep_tree;但在实际生产环境中我对这套语法保持谨慎态度。一方面是因为很多企业的 Hive 还是 3.x 版本另一方面是原生递归 CTE 在底层执行时容易出现性能抖动资源消耗不太好预估。小数据量验证逻辑没问题一旦上了十亿级节点还是老老实实用迭代方案稳妥。2.3 临时表 循环迭代方案这就是 2.0 的核心路线。思路是用一张“结果总表”保存已经找到的所有节点再用一张“增量表”保存本轮新发现的节点每轮迭代从增量表出发向外扩展用总表做去重和防环。这个方案的优势很明显每一轮只扫增量数据而不是全量数据性能可控深度上限可以人为配置逻辑完全透明每轮跑了多少数据一目了然。缺点是需要写 Shell 脚本来控制循环对开发者的工程能力有一定要求。2.4 自定义 UDF / Transform 脚本方案还可以用 Python 脚本配合TRANSFORM语法在脚本内部实现递归逻辑。这种方式逻辑自由度最高但受限于单机内存和脚本执行模型数据量大时容易成为瓶颈。我一般只在“递归内部有复杂业务规则”时才会考虑它普通展开场景用循环迭代就够了。下面用一张表把这几种路线的特点汇总一下方案深度上限实现复杂度性能表现适用场景写死层数自连接固定层数低层数越多越差浅层级、需求稳定原生 WITH RECURSIVE无低受引擎影响大小数据量、Hive 4.0临时表循环迭代可配置中可控需调优企业级大规模数据Python UDF 递归受单机内存限制高中低小数据量、复杂规则我最终选择的是“临时表 循环迭代”下面详细讲它的设计逻辑。3. 2.0 方案核心设计双表迭代框架3.1 结果总表 增量表的协同工作方式整套方案的核心是两个表结果总表TOTAL和增量表INC。每一轮的迭代动作可以做如下拆解从上一轮的增量表中取数据去关联子节点。关联出来的新节点和结果总表做LEFT JOIN如果已经存在则过滤掉。过滤后的数据写入新的增量表。将新增量合并进结果总表。如果新增量为空或者达到了设定的最大迭代次数循环终止。这个流程里最关键的优化点在于每轮迭代的输入是“上一轮的增量”而不是“历史全量”。这样一来数据增长不是随层数线性叠加的整体扫描量小很多这正是 2.0 比 1.0 性能好的根本原因。3.2 环路检测与终止条件设计防环路是递归计算里最不能忽略的一环。环路指的是数据里出现了 A→B→C→A 这种循环引用。在双表迭代的设计下环路其实已经天然被结果总表挡住了任何节点只要被访问过再次出现时会被总表过滤掉。所以迭代不会因为环路而无限循环最多遍历完全部节点就自然会停下来。不过为了更早地剪枝节省无谓的 IO还可以在迭代 SQL 里加一个路径字段用find_in_set判断当前节点是否已经出现在路径上。这种做法的代价是每行都要携带一段路径字符串层级很深时字符串会很长影响性能。我的经验是一般组织架构类数据层级小于 20 层用路径检测没问题如果是 BOM 展开这种层级可能上百的场景仅靠总表过滤就够了路径检测不做也罢。下面是我在实际脚本里用的两个终止条件本轮增量为空说明已经递归到底了正常结束。迭代次数达到预设的MAX_LOOP说明可能存在异常数据比如深层次环强制退出并告警。3.3 数据倾斜与发散节点的处理思路递归迭代中最容易出性能问题的是“发散节点”。举个例子某个根部门下面挂了十万个叶子部门那某一轮迭代中这十万条子节点记录会全部由这一个根部门产生。在 Hive 的分布式计算模型下这意味着某一个 Reduce 要处理巨大的数据量跑得比其它所有 Reduce 都慢也就是典型的数据倾斜。处理思路有几种对关联键做加盐处理把大节点的数据打散到多个 Reduce。拆根并行。如果业务允许把不同的根节点分组拆成多个独立任务并行跑。如果增量表和关联表都不大比如小于 2GB直接加MAPJOIN提示把关联表加载到内存里绕开 Shuffle。另外还有一个容易忽略的问题脏数据导致的“假发散”。如果源表里存在大量parent_dep_id为空字符串或特殊字符的孤儿节点它们会在某一轮被同时关联出来造成一轮迭代的数据量暴涨。建表时对这种情况要有心理准备清洗数据的时候要一并处理。4. 实操全流程完整跑通一个部门树递归展开下面用一个经典场景——部门树展开——走一遍完整流程。假设有一张部门关系表ods_dep_rel我把它简化成下面这种结构CREATE TABLE IF NOT EXISTS ods_dep_rel ( dep_id STRING COMMENT 部门id, parent_dep_id STRING COMMENT 上级部门id, dep_name STRING COMMENT 部门名称, project_id STRING COMMENT 业务线标识 ) COMMENT 部门关系原始表 PARTITIONED BY (dt STRING COMMENT 数据日期) STORED AS ORC;需求非常明确给定业务线project_id找出每个根部门下所有层级的子部门并记录每个子部门相对于根部门的层级深度。4.1 初始化 SQL生成第 0 层数据第 0 层就是根节点。实际业务里根节点的定义可能不同有的是parent_dep_id IS NULL有的是parent_dep_id 还有的是设置了专门的根标识字段这里以最常见的parent_dep_id IS NULL为例。-- init_dep_tree.sql INSERT OVERWRITE TABLE tmp_dep_tree_prev SELECT dep_id AS root_dep_id, dep_id AS dep_id, dep_name AS dep_name, 0 AS loop_id, cast(dep_id AS STRING) AS path_str FROM ods_dep_rel WHERE dt ${hiveconf:dt} AND project_id ${hiveconf:project_id} AND parent_dep_id IS NULL;这轮产出的tmp_dep_tree_prev表既是初始结果集也是第一轮迭代的增量输入。-- 创建结果总表如果不存在 CREATE TABLE IF NOT EXISTS tmp_dep_tree_total ( root_dep_id STRING, dep_id STRING, dep_name STRING, loop_id INT, path_str STRING ) STORED AS ORC;然后把初始数据写入总表INSERT INTO TABLE tmp_dep_tree_total SELECT * FROM tmp_dep_tree_prev;注意这里用的是INSERT INTO因为总表是逐轮追加的。4.2 单轮迭代 SQL增量展开的核心逻辑每一轮迭代的核心 SQL 是同一套逻辑只是输入表不同。落到脚本里我写成参数化的 SQL 文件-- expand_once.sql -- 输入tmp_dep_tree_prev上一轮增量 -- 输出tmp_dep_tree_cur本轮新增量 INSERT OVERWRITE TABLE tmp_dep_tree_cur SELECT t.root_dep_id AS root_dep_id, c.dep_id AS dep_id, c.dep_name AS dep_name, t.loop_id 1 AS loop_id, concat_ws(,, t.path_str, c.dep_id) AS path_str FROM tmp_dep_tree_prev t JOIN ods_dep_rel c ON t.dep_id c.parent_dep_id LEFT JOIN tmp_dep_tree_total total ON c.dep_id total.dep_id AND t.root_dep_id total.root_dep_id WHERE c.dt ${hiveconf:dt} AND c.project_id ${hiveconf:project_id} AND total.dep_id IS NULL AND NOT find_in_set(c.dep_id, t.path_str) 0;这段 SQL 里最核心的是三个地方JOIN以上一轮增量表的dep_id去匹配源表的parent_dep_id找到所有直接子节点。LEFT JOIN ... WHERE total.dep_id IS NULL过滤掉已经在结果总表里出现过的节点这就是上面说的防环路、防重复。find_in_set(c.dep_id, t.path_str)从路径上再做一次检查防止环路在极深路径下带来额外的重复计算。每个细节都很重要。LEFT JOIN的过滤条件必须同时带root_dep_id否则不同根节点下出现同一个部门 ID 时会被错误过滤掉导致结果缺失。4.3 Shell 循环控制脚本让迭代自动收敛有了初始化 SQL 和单轮迭代 SQL剩下就是把循环串起来。我用的是 Bash 脚本 hive -f的方式每一轮迭代完成后检查增量条数决定继续还是终止。#!/bin/bash # 递归计算 2.0部门树自动展开 set -e DT2024-01-01 PROJECT_IDP001 MAX_LOOP50 CUR_LOOP0 HIVE_CONF-hiveconf dt${DT} -hiveconf project_id${PROJECT_ID} # 1. 初始化第 0 层 hive ${HIVE_CONF} -f ./init_dep_tree.sql hive ${HIVE_CONF} -e INSERT INTO TABLE tmp_dep_tree_total SELECT * FROM tmp_dep_tree_prev; # 2. 循环迭代 while [ ${CUR_LOOP} -lt ${MAX_LOOP} ]; do echo Loop: ${CUR_LOOP} # 执行单轮扩展 hive ${HIVE_CONF} -f ./expand_once.sql # 统计本轮增量行数 INC_CNT$(hive -S ${HIVE_CONF} -e SELECT count(1) FROM tmp_dep_tree_cur; | tr -d ) echo Increment count: ${INC_CNT} if [ ${INC_CNT} -eq 0 ]; then echo No new increment, recursion finished. break fi # 合并增量到结果总表 hive ${HIVE_CONF} -e INSERT INTO TABLE tmp_dep_tree_total SELECT * FROM tmp_dep_tree_cur; # 轮转表当前增量变为下一轮的输入 hive ${HIVE_CONF} -e INSERT OVERWRITE TABLE tmp_dep_tree_prev SELECT * FROM tmp_dep_tree_cur; CUR_LOOP$((CUR_LOOP 1)) done # 3. 输出最终结果 hive ${HIVE_CONF} -e SELECT root_dep_id, dep_id, dep_name, loop_id FROM tmp_dep_tree_total ORDER BY root_dep_id, loop_id; 这段脚本有三个经验点值得分享tr -d 是必要的。hive -S -e输出的count结果往往带一些空白字符不清理直接做数值比较会出问题。用INSERT OVERWRITE TABLE tmp_dep_tree_prev SELECT * FROM tmp_dep_tree_cur做表切换而不是TRUNCATE后原地插入是为了保持上一轮数据直到本轮完全成功。一旦中间某一步失败旧的增量数据还在排查和恢复都会容易很多。循环里没必要用break后再单独把结果导出去直接到最后统一输出即可。4.4 结果校验确认递归没有算错递归算完了怎么确认结果对不对我的经验是至少做两层校验。第一层总数校验。用源表跑一个不依赖递归的聚合查询比如每个部门的直接子部门数再和自己算出来的结果做交叉比对。如果源表本身有“该部门所有后代部门数”这样的冗余字段校验会更直接。第二层层级校验。随机挑几个已知深度的部门对照结果表里的loop_id确认深度是否正确。这一步虽然朴素但能有效发现初始化时根节点定义错误导致的全盘偏移。-- 校验示例检查是否有节点的 loop_id 与路径深度不一致 -- 正常情况下 path_str 里逗号的数量加 1 应该等于 loop_id SELECT root_dep_id, dep_id, loop_id, path_str, (LENGTH(path_str) - LENGTH(REPLACE(path_str, ,, )) 1) AS path_depth FROM tmp_dep_tree_total WHERE loop_id ! (LENGTH(path_str) - LENGTH(REPLACE(path_str, ,, )) 1);这个 SQL 利用了逗号计数的技巧很直观。有经验的读者可能已经想到这其实也验证了路径拼接逻辑是否正确。5. 性能调优与踩坑记录2.0 方案里踩过的坑一点也不比业务逻辑少。选几个最有代表性的写出来。5.1 迭代轮次过多导致的小文件问题每一轮迭代都会生成一批新的 ORC 文件如果最大迭代深度是 30 层结果总表里会累积很多小文件。时间久了总表在做LEFT JOIN过滤时甚至会因为元数据膨胀而变慢。我的处理方式是在脚本结束后对结果总表做一次合并压缩。INSERT OVERWRITE TABLE tmp_dep_tree_total SELECT * FROM tmp_dep_tree_total DISTRIBUTE BY root_dep_id;DISTRIBUTE BY root_dep_id会让同一个根节点的数据落到同一个 Reduce不仅合并了小文件还能让后续查询按根节点过滤时更快。5.2 大增量 join 时的任务卡顿当某一轮增量特别大比如某个中间层节点数量暴涨时JOIN过程容易出现长尾。我踩过最厉害的一次是在第 4 层迭代时单个 Reduce 跑到 3 小时没结束。排查之后发现是数据倾斜一个超大父节点带了几十万条子节点记录全部路由到了同一个 Reduce。解决方案是在迭代 SQL 的JOIN关联键上做前缀加盐或者直接把源表按parent_dep_id分桶。分桶后JOIN的 Shuffle 数据量会小很多实测性能提升非常明显。不过加盐方案会改变关联逻辑对非倾斜节点来说反而增加计算量。我的建议是先通过COUNT按parent_dep_id聚合确认倾斜节点的量级再决定是否加盐不要盲目优化。5.3 分区参数穿透导致的串数问题这个坑很隐蔽。在源表ods_dep_rel存在多分区的情况下如果迭代 SQL 里忘记写c.dt ${hiveconf:dt}那么JOIN会扫描全分区。某一次我切换日期参数后忘了清理老的临时表结果递归结果里混了上一个日期的历史数据排查了很久。从那以后我养成了一个习惯所有迭代 SQL 的WHERE条件里源表分区条件必须写全无论哪个环境执行都能保证只读当天数据。5.4 常见报错速查表报错信息或现象可能原因解决办法迭代一直不停止达到 MAX_LOOP存在深层次环路且总表防环逻辑有问题检查LEFT JOIN的关联条件是否带了root_dep_id检查源数据是否有自关联某一轮增量数据暴涨脏数据导致发散节点或孤儿节点大量挂载清洗源数据剔除parent_dep_id为空的孤儿记录对关联键做分桶结果总表小文件过多每轮迭代产生新文件未合并结束后用INSERT OVERWRITE ... DISTRIBUTE BY root_dep_id合并count(1)判断不准导致提前退出hive -S -e返回字符串带空格用tr -d 清理后再做比较不同根节点下相同部门被过滤LEFT JOIN条件漏掉root_dep_id补上AND t.root_dep_id total.root_dep_id定位不到数据问题临时表太多搞不清哪张是上一轮数据固定使用prev/cur轮转命名不要动态拼接表名6. 这套方案还能怎么扩展如果只是做部门树展开这套脚本已经够用了。但递归计算的场景远不止组织架构。这里列几个我实际做过的扩展方向。6.1 BOM 物料多级展开BOM 展开和部门树最大的区别在于子节点关联父节点时还带一个“用量”字段递归展开时要把路径上的用量做乘积最终得到每一个末级物料的累计需求量。实现时只需要在迭代 SQL 的SELECT子句里多带一个用量字段合并时做累乘即可。防环和终止逻辑完全复用。6.2 关系图谱中的多度人脉查找比如“找出某个用户 3 度以内的所有关联用户”。这个场景本质上就是限制深度的递归查找只需要在循环控制脚本里把MAX_LOOP设为 3并且在初始化 SQL 里固定好起始用户即可。6.3 数据血缘追踪数据仓库的表和字段之间存在“上游依赖”关系想查某个指标字段的完整血缘链路可以用同样的迭代框架把“父节点”换成“上游字段”把“子节点”换成“下游字段”。血缘数据的层级往往很深而且环路的可能性比组织架构大不少这时候双表防环机制的可靠性就显得格外重要。最后说点实在的这套 2.0 方案做完之后我再也没有因为“递归查不出来”被业务方找过。回看整个优化过程最值得投入精力的其实不是循环脚本本身而是对增量驱动和防环机制的理解。很多做数据开发的朋友一听到 Hive 递归就想着去装插件、换引擎其实标准 Hive 的临时表迭代方案处理绝大多数生产需求都足够了。如果看完这篇文章你只想记住一个点那我建议记住这句迭代 增量表 结果总表过滤是 Hive 里做递归的通用范式。这个范式一旦建立起来不管遇到组织架构、BOM 展开还是关系图谱都能快速套用最多就是改改表结构、调调终止条件而已。下一次再有人跟你说“Hive 做不了递归”你可以把这篇文章甩给他。
_kaic)
_kaic)
_kaic)
