80TB Hive表Spark查询优化求助：耗时23分钟/13万+任务

1. 强制小表广播，避免大表Shuffle #

input_parameters仅1-10条记录，直接让Spark把小表广播到所有Executor节点，彻底消除大表的Shuffle开销。可以用Hive提示语法，或者开启自动广播配置：

-- 全局开启自动广播（阈值设为10MB，足够覆盖小表）
SET spark.sql.autoBroadcastJoinThreshold=10485760;

-- 或者在查询中显式指定广播小表
SELECT DISTINCT
/*+ BROADCAST(b) */
b.batch_id,
...

2. 合并过滤条件，提前削减大表数据量 #

原查询里子查询a先过滤了trans_dt最近180天，关联后又过滤auth_date（注意到auth_date是trans_dt的别名），可以把日期过滤逻辑合并，同时结合小表的日期范围，在读取大表时就过滤掉无关数据：

WITH input_params AS (
    SELECT batch_id, sid, cust_id, acc_no, credit_13,
           -- 提前计算好小表的日期边界，避免关联时重复计算
           NVL(from_date_filter, DATE_SUB(current_date(),730)) as from_date,
           NVL(to_date_filter, current_date()) as to_date
    FROM input_parameters
)
SELECT DISTINCT
/*+ BROADCAST(input_params) */
ip.batch_id, ip.sid, ip.cust_id, ip.acc_no,
cas.debit_11, cas.credit_13, ...
FROM clouddb.transaction cas
JOIN input_params ip ON cas.credit_13 = ip.credit_13
-- 合并两层日期过滤，只读取真正需要的区间
WHERE cas.trans_dt BETWEEN GREATEST(DATE_SUB(current_date(),180), ip.from_date) 
                       AND LEAST(current_date(), ip.to_date);

3. 移除/优化不必要的DISTINCT #

先确认是否真的需要DISTINCT：

• 如果是小表input_parameters的credit_13有重复导致关联后数据重复，先给小表去重：

  WITH input_params AS (
      SELECT DISTINCT batch_id, sid, cust_id, acc_no, credit_13,
             NVL(from_date_filter, DATE_SUB(current_date(),730)) as from_date,
             NVL(to_date_filter, current_date()) as to_date
      FROM input_parameters
  )
  

• 如果是大表本身有重复数据，用GROUP BY替代DISTINCT，Spark对GROUP BY的执行计划优化通常更高效。

4. 调整Shuffle分区数，减少任务调度开销 #

13万+任务数明显过多，会大幅增加调度耗时。根据集群CPU核数调整Shuffle分区数，比如每核分配2-4个分区：

-- 示例：集群有500核，设置1000个分区
SET spark.sql.shuffle.partitions=1000;

5. 简化嵌套子查询，用CTE提升优化效率 #

把原查询的嵌套子查询改成CTE（公共表表达式），让Spark优化器更容易识别执行计划的优化点，同时提升代码可读性：

WITH input_params AS (
    SELECT batch_id, sid, cust_id, acc_no, credit_13,
           NVL(from_date_filter, DATE_SUB(current_date(),730)) as from_date,
           NVL(to_date_filter, current_date()) as to_date
    FROM input_parameters
),
transaction_data AS (
    SELECT 
        cas.debit_11, cas.credit_13, cas.debit_15, cas.amount, cas.conversion_amount, cas.curr_cd,
        -- 把重复判断逻辑提取成临时字段，减少计算量
        CASE WHEN cas.appr_deny_cd in ('0','1','6') THEN 1 ELSE 0 END as is_approved,
        CASE WHEN cas.appr_deny_cd in ('0','1','6') THEN 'Approved' 
             WHEN cas.appr_deny_cd = '2' THEN 'System Denied' 
             WHEN cas.appr_deny_cd = '3' THEN 'Authorizer Denied' 
             WHEN cas.appr_deny_cd = '4' THEN 'System Pending' 
             WHEN cas.appr_deny_cd = '5' THEN 'Auth Pending' 
             WHEN cas.appr_deny_cd = '7' THEN 'Denied' 
             WHEN cas.appr_deny_cd = '8' THEN 'Pending' 
             WHEN cas.appr_deny_cd = '9' THEN 'Timeout - Reject' 
             ELSE cas.appr_deny_cd END as approval_deny_cd,
        CASE WHEN is_approved = 1 then 'approved' ELSE 'declined' END as approval,
        cas.sed10, cas.sed_pkey, cas.time_of_day_in, cas.trans_dt as auth_date,
        cas.atm_terminal_id, cas.atm_location_addr, cas.atm_street_addr, cas.atm_city_nm,
        cas.atm_state_cd, cas.atm_country_cd, cas.atm_zip_cd, cas.atm_country,
        cas.trx_1, cas.trx_2, cas.trx_3, cas.trx_4, cas.trx_5, cas.trx_6, cas.trx_7, cas.trx_8
    FROM clouddb.transaction cas
    WHERE cas.trans_dt BETWEEN DATE_SUB(current_date(),180) AND current_date()
)
SELECT DISTINCT
/*+ BROADCAST(input_params) */
ip.batch_id, ip.sid, ip.cust_id, ip.acc_no,
td.debit_11, td.credit_13, td.debit_15, td.amount, td.conversion_amount, td.curr_cd,
td.approval_deny_cd, td.approval, td.sed10, td.sed_pkey, td.time_of_day_in, td.auth_date,
td.atm_terminal_id, td.atm_location_addr, td.atm_street_addr, td.atm_city_nm,
td.atm_state_cd, td.atm_country_cd, td.atm_zip_cd, td.atm_country,
td.trx_1, td.trx_2, td.trx_3, td.trx_4, td.trx_5, td.trx_6, td.trx_7, td.trx_8
FROM transaction_data td
JOIN input_params ip ON td.credit_13 = ip.credit_13
WHERE td.auth_date BETWEEN ip.from_date AND ip.to_date;

6. 检查大表的分区与索引 #

• 如果clouddb.transaction还没按trans_dt分区，立即分区：

  ALTER TABLE clouddb.transaction PARTITIONED BY (trans_dt DATE);
  

  分区后查询只会扫描指定日期区间的分区，直接减少IO量。
• 对于关联字段credit_13，可以创建分桶表，加速关联时的匹配：

  ALTER TABLE clouddb.transaction CLUSTERED BY (credit_13) INTO 100 BUCKETS;
  

内容的提问来源于stack exchange，提问作者SNS