1. 这不是“加个GROUP BY”就能搞定的事：多维聚合中的数据变形真相

你有没有遇到过这样的场景：业务方甩来一张报表需求——“按地区、按产品线、按季度，统计销售额、毛利、复购率，再叠加同比和环比”，你信心满满写完SQL，跑出来一看，维度交叉后行数爆炸，空值遍地，SUM(CASE WHEN...)嵌套三层还漏了某个组合，最后不得不导出Excel手动透视？这不是你SQL没学好，而是你还没真正踩进多维聚合的数据操纵深水区。Part 20 这个标题看着像教科书章节编号，但它直指一个被严重低估的硬核能力：Data Manipulation in Multi-Dimensional Aggregation——在多维聚合语境下，对数据进行有目的、可控制、可复现的主动变形操作。它不是简单的分组求和，而是像捏陶土一样，在“地区×产品×时间”这个三维立方体上，切、削、拉、压、旋转、填充、补洞。核心关键词就三个：多维（Multi-Dimensional）、聚合（Aggregation）、操纵（Manipulation）。这三者缺一不可。少了“多维”，就是单表GROUP BY；少了“聚合”，就是普通JOIN或FILTER；少了“操纵”，那就只是被动展示结果。它解决的是真实商业分析中那个最痛的点：原始聚合结果无法直接服务于下游决策动作——比如销售总监要对比华东区A类新品Q3 vs Q2的毛利结构变化，但数据库里只有按“大区+品类+季度”三级分组的总毛利，没有“结构占比”，没有“变动归因”，更没有“异常标记”。这篇文章就是为你写的，无论你是刚能写COUNT(*)的初级分析师，还是天天和Pandas pivot_table搏斗的数据工程师，只要你需要把聚合结果变成“能用、好用、敢用”的决策燃料，你就必须吃透这一关。我干这行十二年，带过三十多个BI项目，90%以上的返工和口径争议，根源都在这一环没做扎实。

2. 为什么不能只靠SQL GROUP BY？多维聚合操纵的本质与设计逻辑

2.1 多维聚合不是“多加几个字段”，而是一个立方体空间建模

很多人把“按A、B、C分组”理解为SQL里GROUP BY a, b, c这么简单。这是致命误解。真正的多维聚合，本质是在构建一个OLAP立方体（OLAP Cube）。想象一个真实的物理立方体：X轴是地区（华东、华北、华南），Y轴是产品线（硬件、软件、服务），Z轴是时间（2023Q1、2023Q2、2023Q3）。每个顶点（如[华东, 硬件, 2023Q2]）就是一个唯一的单元格（Cell），里面存放着该组合下的聚合值（如销售额=1250万）。这个立方体有三个关键属性：维度（Dimensions）、度量（Measures）、层次（Hierarchies）。维度就是坐标轴（地区、产品、时间）；度量就是你要算的数值（销售额、订单数）；层次则是维度内部的上下级关系（比如“时间”维度包含“年→季度→月→日”的层次，“地区”包含“大区→省→市”的层次）。而“Data Manipulation”的起点，恰恰是从这个立方体模型出发的。你不是在操作一堆扁平的行，而是在操作一个有结构、有关系、有路径的立体空间。所以，当业务说“我要看华东区硬件产品在Q2的销售额，以及它占华东区总销售额的比例”，你脑子里第一个反应不应该是“写个子查询”，而应该是：“我在[地区=华东]这个切片上，沿着[产品线]维度做一次‘下钻（Drill-down）’，然后在[华东]这个子立方体里，计算[硬件]单元格的值占整个[华东]切片所有产品线单元格之和的比例”。这个思维转换，是区分“会写SQL”和“懂多维分析”的分水岭。我见过太多人，用几十行嵌套CASE WHEN硬生生把一个本该用两行MDX或DAX就能表达的“父级占比”逻辑写得面目全非，最后连自己都看不懂。

2.2 “Manipulation”的四大核心动作：切、比、补、标

在立方体模型里，“Manipulation”绝非泛泛而谈，它有四个明确、可落地、可编程的核心动作，每一个都对应着真实业务场景的刚需：

1. 切片（Slicing）与切块（Dicing）：这是最基础也最容易被忽视的动作。“切片”是固定一个维度取值，观察其他维度的变化，比如“固定时间=2023Q2，看各地区、各产品线的销售额”。这对应SQL里的WHERE time = '2023Q2'。“切块”则是同时固定多个维度的取值范围，形成一个子立方体，比如“看华东和华南地区、硬件和软件产品、2023Q2和Q3的数据”。这对应WHERE region IN ('华东','华南') AND product IN ('硬件','软件') AND time IN ('2023Q2','2023Q3')。关键在于，切片/切块不是为了过滤数据，而是为了定义后续所有操纵动作的作用域。很多错误就源于作用域没定清楚，比如想算“华东区各产品的Q2占比”，却忘了先切片到华东区，导致分母变成了全国总额。

2. 比较（Comparison）：这是驱动决策的核心。它远不止于“本期vs上期”。在多维空间里，比较有严格的路径依赖：

   • 时间序列比较：同维度组合下，不同时间点的度量值对比（Q2 vs Q1）。这要求时间维度有明确的层次和顺序。
   • 结构占比比较：同一时间点下，某维度成员占其父级或全集的比例（硬件销售额 / 华东区总销售额）。这要求能动态识别“父级”。
   • 对标比较：与指定基准（如行业均值、目标值、标杆区域）的差异（华东Q2实际/目标=112%）。这要求外部基准能与立方体坐标对齐。
   • 变动归因比较：分解总量变动的原因（Q2比Q1增长1000万，其中地区贡献+300万，产品贡献+500万，价格贡献+200万）。这需要更高级的加权计算。

3. 填充与补全（Filling & Completing）：现实世界的数据是稀疏的。一个新成立的“西南区”在2023Q1可能没有任何销售记录，数据库里就根本没有[西南, *, 2023Q1]这条记录。但报表里必须显示“0”。这就是稀疏立方体（Sparse Cube）的稠密化（Densification）。Manipulation必须能主动“造出”那些本不存在的单元格，并赋予合理默认值（0、NULL、前向填充值等）。这在SQL里需要CROSS JOIN生成所有可能组合，再LEFT JOIN原数据，过程繁琐且易错。在Pandas里，reindex()配合fill_value是标准解法，但前提是索引必须是MultiIndex且层级对齐。

4. 标注与标记（Annotation & Flagging）：这是让数据“开口说话”的关键一步。聚合结果本身是冰冷的数字，Manipulation要给它打上业务语义标签：

   • 异常标记：CASE WHEN sales > LAG(sales) OVER (PARTITION BY region, product ORDER BY time) * 1.5 THEN '暴增' ELSE '正常' END
   • 状态标记：根据多个度量组合判断（毛利>0 AND 复购率<30% → “高毛利低粘性”）
   • 等级标记：将连续数值映射为离散等级（销售额0-100万→C级，100-500万→B级，500万+→A级）

这四大动作，构成了Manipulation的完整能力图谱。任何脱离这个图谱的“技巧”，比如单纯用UNION ALL拼接不同粒度的结果，都是在用蛮力绕开问题本质，后患无穷。

2.3 方案选型：为什么放弃纯SQL，拥抱混合技术栈？

面对上述复杂需求，纯SQL方案很快就会崩溃。原因很实在：SQL是为描述“静态集合”而生，不是为描述“动态空间操作”而生。当你需要在一个查询里同时完成“切片→填充→计算父级占比→打异常标”时，SQL的语法会变得极其臃肿。我曾维护过一个报表，其核心SQL长达800行，包含7层嵌套子查询、12个窗口函数、3个CROSS JOIN，每次业务方加一个新维度，整个查询就要重写。这不是技术问题，是范式错配。

因此，我们团队在Part 20实践中，坚定采用混合技术栈（Hybrid Stack）：

• 底层（数据准备层）：用SQL（或Spark SQL）完成最耗资源的原始聚合（GROUP BY region, product, time）和基础切片。目标是产出一个结构清晰、无冗余、已去重的“宽表基底”，比如fact_sales_agg，字段为region_id, product_id, quarter, sales_amt, gross_profit。
• 中层（操纵层）：用Python Pandas（或R dplyr）作为主力操纵引擎。理由非常硬核：Pandas的DataFrame天然支持MultiIndex，pivot_table()、stack()/unstack()、groupby().apply()、reindex()等方法，是对立方体进行“切、比、补、标”的完美抽象。一行df.groupby(['region', 'product']).sales_amt.transform('sum')就能拿到每个组合的“区域-产品”粒度销售额，再除以df.groupby('region').sales_amt.transform('sum')，父级占比瞬间完成。代码可读性、可维护性、可测试性远超等效SQL。
• 上层（呈现层）：用BI工具（如Tableau、Power BI）或自研Web应用消费中层输出的结构化结果。BI工具强大的可视化和交互能力，正好承接Manipulation后的丰富语义。

这个分层不是为了炫技，而是基于每个工具的比较优势（Comparative Advantage）。SQL擅长海量数据的并行扫描和聚合；Pandas擅长灵活、迭代、面向对象的数据变形；BI工具擅长用户交互和视觉传达。强行让一个工具干所有活，就像让挖掘机去绣花，效率低下且效果糟糕。我亲眼见过一个团队，为了“统一技术栈”，坚持用纯SQL实现所有Manipulation，结果开发周期延长3倍，上线后性能下降70%，最终不得不推倒重来。

3. 核心细节拆解：从原始聚合到决策就绪数据的七步实操

3.1 第一步：定义清晰、无歧义的维度与度量字典

一切Manipulation的起点，是一份被所有人认可的、版本化的数据字典。这不是文档工作，是技术基建。我们为Part 20项目制定了如下核心字典：

提示：维度ID必须是稳定、无业务含义的代理键（Surrogate Key）。我吃过亏：早期用region_name作为分组字段，结果市场部把“华东区”改名为“华东及华中大区”，所有历史报表口径瞬间失效。现在一律用region_id，名字变更只影响维度表的name字段，不影响事实表关联和聚合逻辑。

度量（Measures）同样需要明确定义：

• sales_amt：确认收入，单位：人民币元，精度：2位小数，不含税。
• gross_profit：sales_amt-cost_of_goods_sold，必须与sales_amt在同一会计期间确认。
• order_count：去重订单数，DISTINCT order_id，非行数。

这份字典，是我们所有Manipulation操作的“宪法”。任何对聚合逻辑的讨论，都必须回归到字典定义。它杜绝了“我以为的销售额”和“你理解的销售额”之间的扯皮。

3.2 第二步：构建“最小完备”原始聚合宽表

不要试图在第一步就产出所有维度组合。我们的原则是：只聚合到业务逻辑上最细、最不可再分的粒度（Atomic Granularity）。对于销售分析，这个粒度通常是[region_id, product_id, quarter_id]。我们用Spark SQL执行：

-- Part 20: Base Aggregation Query INSERT OVERWRITE TABLE fact_sales_agg SELECT COALESCE(r.region_id, 0) AS region_id, COALESCE(p.product_id, 'UNKNOWN') AS product_id, CONCAT(YEAR(o.order_date), 'Q', QUARTER(o.order_date)) AS quarter_id, SUM(o.sales_amount) AS sales_amt, SUM(o.sales_amount - o.cost_amount) AS gross_profit, COUNT(DISTINCT o.order_id) AS order_count FROM ods_orders o LEFT JOIN dim_region r ON o.region_code = r.region_code LEFT JOIN dim_product p ON o.product_sku = p.sku_code WHERE o.order_status = 'CONFIRMED' AND o.order_date >= '2022-01-01' GROUP BY COALESCE(r.region_id, 0), COALESCE(p.product_id, 'UNKNOWN'), CONCAT(YEAR(o.order_date), 'Q', QUARTER(o.order_date));

关键细节：

• COALESCE处理NULL：确保维度ID总有值，避免后续GROUP BY产生意外的NULL组合。
• CONCAT生成标准季度ID：统一格式，避免'2023-Q2'、'Q2-2023'等混乱。
• WHERE条件前置：在聚合前过滤掉无效订单，极大提升性能。
• 不JOIN维度表的名称字段：只保留ID。名称、层级等描述性信息，留到Pandas层通过map()或merge()动态注入，保证灵活性。

这张宽表，就是我们Manipulation的唯一数据源。它的行数，就是这个三维立方体中所有“非零”单元格的数量。我们称它为“最小完备”表，因为它包含了所有后续操作所需的原子信息，且没有冗余。

3.3 第三步：Pandas中构建MultiIndex立方体骨架

加载宽表后，第一件事不是计算，而是构建立方体的骨架（Skeleton）。这一步决定了后续所有Manipulation的稳定性和可扩展性。

import pandas as pd import numpy as np # 1. 加载基础聚合表 df_base = spark.read.table("fact_sales_agg").toPandas() # 2. 构建MultiIndex骨架：所有可能的维度组合 # 获取各维度的唯一值列表（含'ALL'汇总码） regions = list(df_base['region_id'].unique()) + ['ALL'] products = list(df_base['product_id'].unique()) + ['ALL'] quarters = list(df_base['quarter_id'].unique()) # 创建完整的笛卡尔积索引 idx_full = pd.MultiIndex.from_product( [regions, products, quarters], names=['region_id', 'product_id', 'quarter_id'] ) # 3. 将基础数据设置为MultiIndex DataFrame df_cube = df_base.set_index(['region_id', 'product_id', 'quarter_id']) # 4. 用full index重新索引，实现"稠密化"，缺失值填0 df_cube = df_cube.reindex(idx_full, fill_value=0).reset_index()

这段代码完成了三件大事：

• from_product：生成了[region_id, product_id, quarter_id]所有可能的组合，包括['ALL', 'ALL', '2023Q2']这种全汇总。
• set_index+reindex：将稀疏的原始数据，“拉伸”成一个稠密的立方体骨架。原来没有[5, 'PROD-001', '2023Q1']记录？没关系，reindex会自动创建这一行，sales_amt等字段填0。
• reset_index：把MultiIndex变回普通列，方便后续groupby等操作。

注意：fill_value=0是安全的，因为我们处理的是销售额、订单数等可加性度量（Additive Measures）。如果是平均值、比率等半可加性（Semi-additive）或不可加性（Non-additive）度量，这里就不能简单填0，而要用np.nan，并在后续计算中显式处理缺失值。这是新手常踩的坑——用0填充毛利率，导致“无销售区域”的毛利率被算成0%，严重误导决策。

3.4 第四步：执行核心Manipulation——“切、比、补、标”

现在，立方体骨架已就位，我们开始注入灵魂。以下代码，全部基于df_cube这个稠密DataFrame：

# 1. 【切片】：固定时间，分析Q2 df_q2 = df_cube[df_cube['quarter_id'] == '2023Q2'].copy() # 2. 【比较-结构占比】：计算各产品线在各地区的销售额占比 # 步骤a：先计算每个地区的总销售额（父级聚合） region_total = df_q2.groupby('region_id')['sales_amt'].transform('sum') # 步骤b：计算占比，注意处理分母为0 df_q2['sales_pct_in_region'] = np.where( region_total > 0, df_q2['sales_amt'] / region_total, 0.0 ) # 3. 【比较-时间序列】：计算Q2 vs Q1的环比增长率 # 先获取Q1数据 df_q1 = df_cube[df_cube['quarter_id'] == '2023Q1'].copy() # 重命名Q1的度量列，便于合并 df_q1 = df_q1.rename(columns={ 'sales_amt': 'sales_amt_q1', 'gross_profit': 'gross_profit_q1', 'order_count': 'order_count_q1' }) # 与Q2数据按region_id, product_id合并 df_compare = pd.merge( df_q2, df_q1[['region_id', 'product_id', 'sales_amt_q1', 'gross_profit_q1', 'order_count_q1']], on=['region_id', 'product_id'], how='left' ) # 计算环比（处理Q1为0的情况） df_compare['sales_mom_pct'] = np.where( df_compare['sales_amt_q1'] > 0, (df_compare['sales_amt'] - df_compare['sales_amt_q1']) / df_compare['sales_amt_q1'], np.nan # Q1无数据，环比无意义 ) # 4. 【标注-异常标记】：标记销售额环比增长>50%的产品 df_compare['sales_mom_flag'] = np.where( df_compare['sales_mom_pct'] > 0.5, '暴增', np.where(df_compare['sales_mom_pct'] < -0.3, '暴跌', '平稳') ) # 5. 【补全-父级汇总】：添加'ALL'维度的汇总行 # 按region_id='ALL'汇总所有产品和季度 all_region = df_cube.groupby(['product_id', 'quarter_id'])[ ['sales_amt', 'gross_profit', 'order_count'] ].sum().reset_index() all_region['region_id'] = 'ALL' # 按product_id='ALL'汇总所有地区和季度 all_product = df_cube.groupby(['region_id', 'quarter_id'])[ ['sales_amt', 'gross_profit', 'order_count'] ].sum().reset_index() all_product['product_id'] = 'ALL' # 合并所有基础数据和汇总数据 df_final = pd.concat([ df_compare, all_region, all_product ], ignore_index=True)

这段代码展示了Manipulation的威力：

• 结构占比：用transform('sum')在分组内广播父级总和，一行代码搞定，无需子查询。
• 时间序列比较：用merge替代SQL的JOIN，逻辑清晰，且how='left'保证了Q2有数据而Q1无数据的组合不会丢失。
• 异常标注：np.where链式调用，语义一目了然。
• 父级汇总：groupby().sum()后concat，轻松生成'ALL'行，为BI工具的“钻取”功能提供数据基础。

3.5 第五步：处理半可加性度量——毛利率的特殊战术

毛利率（gross_profit / sales_amt）是典型的半可加性度量：它不能跨时间相加（Q1毛利率+Q2毛利率≠H1毛利率），也不能跨地区相加。直接对gross_profit和sales_amt分别求和再相除，是常见错误。

正确做法是：在最细粒度上计算，再按需聚合。我们在Part 20中采用了“分子分母分离，聚合后计算”的策略：

# 在df_base（原始聚合）上，先计算每行的毛利率（避免0除） df_base['gp_rate_raw'] = np.where( df_base['sales_amt'] > 0, df_base['gross_profit'] / df_base['sales_amt'], np.nan ) # 当需要计算“华东区Q2的毛利率”时： # 步骤1：取出华东区Q2的所有明细行（即使原始宽表是聚合的，我们也保留了明细ID的线索） # 步骤2：对这些行的gross_profit求和，对sales_amt求和 # 步骤3：用总毛利 / 总销售额 # 这在Pandas中，就是： df_east_q2 = df_base[(df_base['region_id'] == 5) & (df_base['quarter_id'] == '2023Q2')] east_q2_gp_total = df_east_q2['gross_profit'].sum() east_q2_sales_total = df_east_q2['sales_amt'].sum() east_q2_gp_rate = east_q2_gp_total / east_q2_sales_total if east_q2_sales_total > 0 else np.nan # 如果必须在宽表上操作，就用'weighted average'近似： # (sum(gp_rate_raw * sales_amt) / sum(sales_amt)) df_cube['weighted_gp_rate'] = df_cube['gp_rate_raw'] * df_cube['sales_amt'] # 然后在groupby时，用sum(weighted_gp_rate) / sum(sales_amt)

这个例子说明，Manipulation不是一套死板的公式，而是需要深入理解业务度量的数学性质。没有银弹，只有针对不同度量的定制化战术。

4. 实操过程全记录：从环境搭建到生产部署的完整流水线

4.1 环境与依赖：轻量、稳定、可复现

我们摒弃了复杂的虚拟环境管理，采用最务实的方案：

• Python版本：3.9.16（LTS长期支持版，避免新特性带来的兼容性风险）
• 核心库：
  • pandas==1.5.3（1.5.x系列是最后一个支持Python 3.7+且API稳定的版本，避免1.4.x的bug和2.x的breaking change）
  • numpy==1.23.5（与pandas 1.5.3完美匹配）
  • pyarrow==11.0.0（大幅提升Pandas读写Parquet的性能，尤其对大宽表）
• 部署方式：所有代码和依赖打包为一个>FROM python:3.9.16-slim # 安装系统依赖 RUN apt-get update && apt-get install -y \ libglib2.0-0 \ libsm6 \ libxext6 \ && rm -rf /var/lib/apt/lists/* # 复制并安装Python依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制代码 COPY src/ /app/src/ WORKDIR /app # 入口脚本 CMD ["python", "src/main.py"]

  requirements.txt内容精简到极致：

  pandas==1.5.3 numpy==1.23.5 pyarrow==11.0.0 pyspark==3.4.0 # 用于连接Spark集群读取宽表

  实操心得：我们曾经在测试环境用pandas==2.0.0，结果发现reindex()在处理超大MultiIndex时内存暴涨300%，且transform()行为有细微变化，导致一个关键报表的占比计算偏差0.02%。线上环境绝对不用非LTS版本。稳定压倒一切。

  4.2 核心脚本main.py：一个文件，五个阶段

  整个Manipulation流程被封装在一个清晰的main.py中，分为五个阶段，每个阶段都有明确的输入、输出和日志：

  # src/main.py import logging from datetime import datetime import pandas as pd import numpy as np from pyspark.sql import SparkSession # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/var/log/manipulation/part20.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) def stage_1_load_data(spark: SparkSession) -> pd.DataFrame: """阶段1：从Spark读取基础聚合宽表""" logger.info("Stage 1: Loading base aggregation table...") df = spark.read.table("fact_sales_agg").toPandas() logger.info(f"Loaded {len(df)} rows from fact_sales_agg") return df def stage_2_build_cube(df_base: pd.DataFrame) -> pd.DataFrame: """阶段2：构建稠密MultiIndex立方体骨架""" logger.info("Stage 2: Building dense cube skeleton...") # ... (同3.3节代码) logger.info(f"Cube skeleton built with {len(df_cube)} rows") return df_cube def stage_3_apply_manipulation(df_cube: pd.DataFrame) -> pd.DataFrame: """阶段3：执行核心Manipulation逻辑""" logger.info("Stage 3: Applying core manipulations (slice, compare, fill, flag)...") # ... (同3.4节代码) logger.info("Core manipulations applied successfully") return df_final def stage_4_validate_output(df_final: pd.DataFrame): """阶段4：关键业务规则校验""" logger.info("Stage 4: Running business rule validations...") # 校验1：所有'sales_pct_in_region'应在[0,1]区间 pct_outliers = df_final[ (df_final['sales_pct_in_region'] < 0) | (df_final['sales_pct_in_region'] > 1) ] if len(pct_outliers) > 0: raise ValueError(f"Found {len(pct_outliers)} rows with invalid sales_pct_in_region") # 校验2：'ALL'汇总行的sales_amt应等于其下属所有行的sum all_region_sum = df_final[df_final['region_id'] == 'ALL']['sales_amt'].sum() non_all_sum = df_final[df_final['region_id'] != 'ALL']['sales_amt'].sum() if abs(all_region_sum - non_all_sum) > 1e-6: raise ValueError("Region 'ALL' total does not match sum of non-ALL regions") logger.info("All business validations passed") def stage_5_save_output(df_final: pd.DataFrame): """阶段5：保存结果到目标表""" logger.info("Stage 5: Saving final result to target table...") # 使用PyArrow高效写入Parquet df_final.to_parquet( "/data/output/part20_final.parquet", engine='pyarrow', compression='snappy', index=False ) logger.info("Final result saved successfully") if __name__ == "__main__": start_time = datetime.now() logger.info(f"Part 20 Data Manipulation STARTED at {start_time}") try: # 初始化Spark spark = SparkSession.builder \ .appName("Part20-Manipulation") \ .getOrCreate() # 执行五个阶段 df_base = stage_1_load_data(spark) df_cube = stage_2_build_cube(df_base) df_final = stage_3_apply_manipulation(df_cube) stage_4_validate_output(df_final) stage_5_save_output(df_final) end_time = datetime.now() duration = end_time - start_time logger.info(f"Part 20 Data Manipulation COMPLETED in {duration.total_seconds():.2f} seconds") except Exception as e: logger.error(f"Part 20 Data Manipulation FAILED: {str(e)}", exc_info=True) raise finally: if 'spark' in locals(): spark.stop()

  这个脚本的设计哲学是：可调试、可监控、可审计。每个阶段都有日志记录起止时间和关键指标；stage_4_validate_output是质量防火墙，任何业务规则违反都会中断流程，绝不让脏数据流入下游；try/except/finally确保Spark资源被正确释放。我们在线上运行时，会将日志实时推送至ELK，任何ERROR级别的日志都会触发企业微信告警。

  4.3 生产调度：Airflow DAG的健壮性设计

  我们使用Apache Airflow 2.6.3进行调度。DAG定义的关键在于失败重试与依赖隔离：

  # dags/part20_dag.py from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.bash import BashOperator from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator from datetime import datetime, timedelta default_args = { 'owner': 'data-engineering', 'depends_on_past': False, 'start_date': datetime(2023, 1, 1), 'email_on_failure': True, 'email_on_retry': False, 'retries': 2, # 仅重试2次，避免无限循环 'retry_delay': timedelta(minutes=5), 'execution_timeout': timedelta(hours=2), # 防止任务卡死 } dag = DAG( 'part20_data_manipulation', default_args=default_args, description='Part 20: Multi-Dimensional Aggregation Manipulation', schedule_interval='0 3 * * 1', # 每周一凌晨3点执行 catchup=False, tags=['data-manipulation', 'sales-analysis'], ) # 任务1：检查上游宽表是否已就绪（防依赖断裂） check_upstream = BashOperator( task_id='check_upstream_table', bash_command='spark-sql -e "SELECT COUNT(*) FROM fact_sales_agg WHERE quarter_id = \'{{ macros.ds_format(ds, \'%Y-%m-%d\', \'%YQ%q\') }}\'" | grep -q "0" && exit 1 || echo "Upstream ready"', dag=dag, ) # 任务2：运行Manipulation主脚本（Docker容器） run_manipulation = BashOperator( task_id='run_manipulation_script', bash_command='docker run --rm -v /data:/data -v /var/log:/var/log>问题现象可能根因排查步骤解决方案报表中出现大量NaN或Inf1. 分母为0的除法运算
  2.np.where条件判断逻辑错误，分支未覆盖所有情况
  3.reindex(fill_value=0)误用于非可加性度量1. 在stage_3_apply_manipulation中，对所有/运算前后加print(f"DEBUG: denom={denom}, num={num}")
  2. 用df_final.isna().sum()和`df_final.is