Hadoop慢查询优化求助：关联T5及终阶段耗时过长

First, let's recap your current bottlenecks: the query runs over 2 hours, with most time spent on joining with table T5 and the final ranking stage. You've already tried converting subqueries to CASE statements and simplifying joins for T3/T4/T6 with limited success, and the ORDER BY logic can't be replaced with aggregations instead of rank().

Here are targeted optimizations to address your specific issues:

1. Eliminate Repeated Subqueries with CTEs #

Your query references SELECT tr_dt FROM DT_LKP three times (for T3, T6, T4). Repeating this subquery forces Hadoop to scan DT_LKP multiple times. Use a Common Table Expression (CTE) to compute this once and reuse it:

WITH dt_lkp AS (
    SELECT tr_dt FROM DT_LKP
)
SELECT * FROM (
    SELECT 
        T2.f1, T2.f2, .... T5.f19, T5.f20,
        -- Precompute the CASE statement once (see optimization 3)
        CASE 
            WHEN T1.trxn_id IS NULL THEN T2.crt_ts 
            WHEN T1.trxn_id IS NOT NULL AND T5.acct_trxn_id IS NOT NULL AND T2.crt_ts >= T5.crt_ts THEN T2.crt_ts 
            WHEN T1.trxn_id IS NOT NULL AND T5.acct_trxn_id IS NOT NULL AND T2.crt_ts < T5.crt_ts THEN T5.crt_ts 
        END AS crt_ts,
        -- Precompute group key for partition (see optimization 3)
        IF(T1.trxn_id IS NULL, 'NULL', T1.trxn_id) AS group_key,
        row_number() over ( 
            PARTITION BY T2.w_trxn_id, group_key 
            ORDER BY T2.business_effective_ts DESC, crt_ts DESC 
        ) AS rnk 
    FROM(
        SELECT f1, f2, w_trxn_id, business_effective_ts, crt_ts 
        FROM T3 
        WHERE title_name = 'CAPTURE' AND tr_dt IN (SELECT tr_dt FROM dt_lkp)
    ) T2 
    LEFT JOIN (
        SELECT trxn_id, w_trxn_id, business_effective_ts 
        FROM T6 
        WHERE tr_dt IN (SELECT tr_dt FROM dt_lkp)
    ) T1 ON T2.w_trxn_id = T1.w_trxn_id AND T2.business_effective_ts = T1.business_effective_ts 
    LEFT JOIN (
        SELECT f1, f3, ..., f20, acct_trxn_id, crt_ts 
        FROM T4 
        WHERE tr_dt IN (SELECT tr_dt FROM dt_lkp)
    ) T5 ON (T1.trxn_id = T5.acct_trxn_id) OR (T1.trxn_id IS NULL AND T5.acct_trxn_id IS NULL)
) FNL 
WHERE rnk = 1

2. Prune Unnecessary Columns & Leverage Partitioning #

• Stop using SELECT *: Only select the exact columns you need for each subquery (as shown in the CTE example above). This reduces data transfer and memory usage across all stages.
• Ensure tr_dt is a partition column: If T3, T4, T6 are partitioned by tr_dt, the filter tr_dt IN (...) will skip scanning irrelevant partitions, drastically reducing the amount of data processed. If they aren't partitioned, consider partitioning these tables by tr_dt for long-term gains.

3. Precompute Repeated Expressions #

Your row_number() window function repeats the same CASE statement used for crt_ts, and the partition uses a repeated IF() expression. Precompute these as separate columns in the inner query:

• This avoids recalculating the same logic hundreds of thousands/millions of times (once per row in the window function).
• Makes the query cleaner and helps the query optimizer recognize opportunities to cache these values.

4. Optimize the T5 Join Condition #

Your original WHERE clause includes:

if(T1.trxn_id is null, 'NULL', T1.trxn_id) = if(T5.acct_trxn_id is null, 'NULL', T5.acct_trxn_id)

This condition effectively filters rows where T1 and T5's transaction IDs are either both NULL or equal. Move this logic directly into the T5 JOIN ON clause (as shown in the CTE example) to filter irrelevant rows earlier in the execution pipeline, reducing the data volume passed to the ranking stage.

5. Address Data Skew (Critical for Long-Running Joins/Rankings) #

The T5 join and window function partition are common points for data skew. Check if:

• Any values in T2.w_trxn_id or group_key have an unusually high number of rows (e.g., a single w_trxn_id with millions of records).
• Use Hive skew optimization parameters:

  set hive.optimize.skewjoin=true;
  set hive.skewjoin.key=100000; -- Adjust based on your data size
  set hive.optimize.skewjoin.compiletime=true;
  

• For extreme skew, split the query into two parts: handle the skewed keys separately, then union the results with the non-skewed data.

6. Update Table Statistics #

Outdated statistics can lead to poor query plans. Refresh statistics for your tables to help the optimizer make better decisions about join orders and resource allocation:

ANALYZE TABLE T3 COMPUTE STATISTICS FOR COLUMNS tr_dt, w_trxn_id, business_effective_ts;
ANALYZE TABLE T4 COMPUTE STATISTICS FOR COLUMNS tr_dt, acct_trxn_id;
ANALYZE TABLE T6 COMPUTE STATISTICS FOR COLUMNS tr_dt, w_trxn_id, business_effective_ts;

7. Simplify the Final Filter with QUALIFY (If Supported) #

If you're using Hive 2.3+ or a Hadoop SQL engine that supports the QUALIFY clause, you can eliminate the outer subquery by filtering directly after the window function:

WITH dt_lkp AS (
    SELECT tr_dt FROM DT_LKP
)
SELECT 
    T2.f1, T2.f2, .... T5.f19, T5.f20,
    CASE 
        WHEN T1.trxn_id IS NULL THEN T2.crt_ts 
        WHEN T1.trxn_id IS NOT NULL AND T5.acct_trxn_id IS NOT NULL AND T2.crt_ts >= T5.crt_ts THEN T2.crt_ts 
        WHEN T1.trxn_id IS NOT NULL AND T5.acct_trxn_id IS NOT NULL AND T2.crt_ts < T5.crt_ts THEN T5.crt_ts 
    END AS crt_ts,
    row_number() over ( 
        PARTITION BY T2.w_trxn_id, IF(T1.trxn_id IS NULL, 'NULL', T1.trxn_id) 
        ORDER BY T2.business_effective_ts DESC, crt_ts DESC 
    ) AS rnk 
FROM(
    SELECT f1, f2, w_trxn_id, business_effective_ts, crt_ts 
    FROM T3 
    WHERE title_name = 'CAPTURE' AND tr_dt IN (SELECT tr_dt FROM dt_lkp)
) T2 
LEFT JOIN (
    SELECT trxn_id, w_trxn_id, business_effective_ts 
    FROM T6 
    WHERE tr_dt IN (SELECT tr_dt FROM dt_lkp)
) T1 ON T2.w_trxn_id = T1.w_trxn_id AND T2.business_effective_ts = T1.business_effective_ts 
LEFT JOIN (
    SELECT f1, f3, ..., f20, acct_trxn_id, crt_ts 
    FROM T4 
    WHERE tr_dt IN (SELECT tr_dt FROM dt_lkp)
) T5 ON (T1.trxn_id = T5.acct_trxn_id) OR (T1.trxn_id IS NULL AND T5.acct_trxn_id IS NULL)
QUALIFY rnk = 1

This reduces query complexity and can help the optimizer streamline execution.

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