多维聚合实战：滚动窗口与业务逻辑嵌入的生产级实现

1. 项目概述为什么多维聚合不是“加个groupby”就能搞定的事我在银行数据平台组干了八年从最早用SQL写几十行嵌套子查询做客户分层到现在每天在Jupyter里调试pandas的agg链式调用踩过的坑比写的代码还多。今天这篇讲的“多维聚合”绝不是教你怎么把df.groupby(col).sum()敲得更顺——那是实习生第一天就能学会的。真正卡住业务分析师、拖慢风控模型上线、让报表系统半夜报警的永远是那些看似简单、实则暗藏玄机的聚合需求比如“请按城市商户类型交易时段统计过去30天内每类客户的平均单笔金额、中位数、标准差同时计算该区间内最大单笔与最小单笔的差值并对每个组合输出滚动7日均值和累计消费总额”。你试试看这一句话里埋了多少个技术雷区我见过太多团队把这类需求拆成七八个独立脚本中间用CSV临时文件接力跑一次要23分钟出错重跑就得从头再来。核心关键词就三个多维聚合、滚动窗口、业务逻辑嵌入。它们共同指向一个现实真实世界的分析场景从来不是单维度切片而是立体交叉从来不是静态快照而是带时间纵深的动态视图更关键的是所有计算必须承载可审计、可复现、可解释的业务规则。比如“高价值交易”的定义在信用卡风控里可能是单笔≥300元且发生在凌晨2点至5点而在跨境支付场景下可能变成单笔≥等值5000美元且收款方注册地为高风险司法管辖区——这种逻辑没法靠np.mean()内置函数解决必须亲手把它焊进聚合流程里。这篇文章适合三类人第一类是刚转行做数据分析的新人正被老板一句“把客户按地区和产品线分组算几个指标”搞得无从下手第二类是已有两年经验的工程师发现现有聚合脚本在数据量涨到千万级后开始卡顿但又说不清瓶颈在哪第三类是业务部门的数据接口人天天被风控、运营、财务轮番追问“为什么这个数和上个月对不上”却找不到聚合逻辑的源头。别急我们不讲虚的直接从银行最真实的信用卡交易分析场景切入把每一步操作背后的“为什么”掰开揉碎——比如为什么unstack()之后必须加fill_value0为什么滚动窗口的min_periods参数设成3而不是1为什么自定义函数里要显式处理空序列。这些细节才是决定你产出能否被业务方签字确认的关键。2. 多维聚合的核心设计思路从“能跑通”到“能扛住生产环境”2.1 为什么拒绝“先group再merge”的老路子很多新手遇到多列聚合需求第一反应是拆解先按商户类型算一次均值再按地区算一次标准差最后用pd.merge()拼起来。这在100行数据里当然没问题但放到银行的真实场景里就是灾难的开始。我去年接手过一个清算系统报表原始脚本就是这么写的——它要聚合全国36个分行、427个支行、18类商户的每日交易数据共1200万行。每次执行pandas会为每个groupby操作单独构建索引、排序、分组三次独立操作下来内存峰值冲到16GBCPU占用率持续98%跑完要47分钟。更糟的是当某次上游数据源漏传了“教育类”商户的记录第一个groupby返回空结果后续merge直接报KeyError整个任务失败但错误日志只显示“合并失败”没人知道是哪个维度缺失了数据。而agg()字典映射方案本质是单次分组、多路并行计算。pandas底层会一次性完成分组索引构建然后将不同列的聚合函数分配到同一分组块上并行执行。就像工厂流水线传送带分组索引只走一遍但上面同时有三台机器mean、median、std在各自加工不同零件不同列。实测对比同样1200万行数据拆解式脚本耗时47分钟内存峰值16GB改用agg({amount:[mean,median],fee:[min,max]})后耗时压到6分12秒内存峰值稳定在3.2GB。这不是语法糖是计算范式的升级。提示当你看到聚合脚本执行时间超过10分钟第一反应不该是加服务器而是检查是否在用多个独立groupby。生产环境里每一次额外的groupby都是对资源的奢侈浪费。2.2 分层列名MultiIndex不是bug是你的数据资产目录运行df.groupby([region,product]).agg({revenue:mean}).unstack()后你看到的输出是product Gadget Widget region North 12000.0 15500.0 South 13750.0 18000.0很多人第一反应是“这列名怎么带括号太丑了赶紧扁平化”——停这个看似麻烦的分层结构恰恰是pandas给你埋下的最强扩展性伏笔。想象一下如果未来业务方要求增加“毛利率”指标你只需在agg()里加一项profit_margin: mean输出自动变成三层列[(revenue,mean), (profit_margin,mean)]。而如果你提前用columns.map(_.join)强行扁平成revenue_mean、profit_margin_mean下次加新指标就得手动改所有下游代码包括BI工具里的字段映射、邮件模板里的变量名、甚至Excel导出的列标题。真正的生产级做法是拥抱分层控制展平时机。在数据处理管道内部全程保持MultiIndex结构直到最后一步导出给业务方时才用result.columns [_.join(col).strip() for col in result.columns.values]统一格式化。这样既保证了中间计算的灵活性又满足了交付物的易读性。我经手的所有银行报表系统都强制要求ETL脚本输出保留原始分层结构BI工程师反馈说他们配置仪表盘时拖拽字段的准确率从73%提升到99%因为再也不用猜“avg_revenue”到底对应的是均值还是中位数了。2.3 “多维”不等于“堆砌维度”维度选择是业务决策groupby([region,product,channel,time_period])看起来很全面但实际中维度越多结果集越稀疏。我处理过一个案例某银行想分析“手机银行APP渠道下华东地区各城市、各年龄段客户的月均交易频次”。他们最初设了4个维度结果生成的DataFrame有21万行其中83%的单元格是NaN——因为很多小城市根本没有足够样本覆盖所有年龄段。最终报表根本无法呈现业务方抱怨“数据太多反而看不到重点”。解决方案不是删维度而是用业务逻辑做预过滤。我们和风控部确认单城市月交易客户数低于500人的不纳入分析年龄段按监管要求必须分5段18-25,26-35,36-45,46-55,55但若某段人数50则合并到相邻段。调整后维度精简为[region,city_group,age_band]city_group是按客户数聚类的三级城市分组结果集压缩到1.2万行有效数据占比91%。记住多维聚合的终极目标不是穷尽所有组合而是找到业务可行动的最小颗粒度。那个“华东-上海-26-35岁”的组合如果能触发精准营销策略它就是黄金维度如果只是躺在报表里吃灰再精细也是资源浪费。3. 核心实操要点从代码片段到生产级实现的七道关卡3.1 多列聚合如何避免“列名冲突”和“数据类型错乱”看这段典型代码result df.groupby(merchant_category).agg({ transaction_amount: [mean,median], processing_fee: [min,max] })表面看没问题但生产环境里transaction_amount列如果是object类型比如混入了字符串“N/A”mean()会直接报错TypeError: unsupported operand type(s) for : float and str。而median()在遇到非数值时会静默返回NaN导致你拿到一个“部分成功”的结果却浑然不觉。实操要点一聚合前必做类型强校验# 在agg前插入校验 def safe_numeric_cast(series, col_name): try: # 尝试转为float强制将非数字转为NaN return pd.to_numeric(series, errorscoerce) except Exception as e: raise ValueError(f列 {col_name} 类型转换失败: {e}) # 对目标列批量处理 for col in [transaction_amount, processing_fee]: df[col] safe_numeric_cast(df[col], col)实操要点二处理分层列名的“命名冲突”当两个不同列都用了mean函数比如{amount:mean, fee:mean}输出列名会是(amount,mean)和(fee,mean)看着没问题。但如果某天你加了count而原始数据里恰好有count列pandas会报ValueError: column count already exists。安全做法是显式指定聚合函数别名result df.groupby(merchant_category).agg({ transaction_amount: [(amt_mean,mean), (amt_median,median)], processing_fee: [(fee_min,min), (fee_max,max)] }) # 输出列名变为amt_mean, amt_median, fee_min, fee_max —— 绝对无歧义3.2 自定义聚合函数为什么lambda只能用于调试不能上生产原文示例中的lambda x: x.max() - x.min()写起来痛快但上线后会出大问题。去年我们有个实时风控模块就因用了lambda函数导致线上服务重启后所有聚合逻辑失效——因为lambda函数无法被pickle序列化而我们的分布式任务调度器Airflow需要序列化函数到工作节点。更隐蔽的问题是lambda没有__name__属性当聚合报错时日志里只显示lambda你根本不知道是哪个业务规则崩了。生产级替代方案带完整契约的命名函数def transaction_range(series, threshold0): 计算交易金额范围最大值-最小值 :param series: pandas.Series交易金额序列 :param threshold: 过滤阈值低于此值的交易视为无效如手续费为0的测试交易 :return: float有效交易的范围值 :raises ValueError: 当有效交易数少于2时抛出异常range需至少2个点 if not isinstance(series, pd.Series): raise TypeError(输入必须是pandas.Series) valid_series series[series threshold] if len(valid_series) 2: raise ValueError(f有效交易数({len(valid_series)})不足2个无法计算range) return valid_series.max() - valid_series.min() # 使用时 result df.groupby(merchant_category).agg({ transaction_amount: [(range, transaction_range)] })这个函数解决了四个生产痛点① 类型检查防止上游数据污染② 业务阈值参数化避免硬编码③ 异常信息明确便于监控告警④ docstring可被Sphinx自动提取生成API文档。我们团队规定所有上生产的自定义聚合函数必须包含类型检查、参数校验、异常说明三要素缺一不可。3.3 滚动窗口为什么window3不是“取最近3条”而是“取时间序最近3条”这是新手最容易栽跟头的地方。看原文代码df_ts[rolling_avg] df_ts.groupby(category)[daily_revenue].rolling(window3).mean()如果df_ts的索引是date且date是datetime64类型那rolling(window3)确实按时间顺序取最近3天。但如果索引是默认的RangeIndex(0,1,2...)或者date列没设为索引rolling()会按物理行序取3行完全无视日期先后。我亲眼见过一个案例某分行导出的交易数据因Excel保存时日期列被转成文本pd.to_datetime()失败后全变NaTrolling()就按行号滚动结果把上周五和本周一的数据强行算在一起风控模型误判了237笔“异常高频交易”。实操铁律滚动窗口前必做三件事确认索引是时间类型assert isinstance(df.index, pd.DatetimeIndex), 索引必须是DatetimeIndex检查数据是否按时间排序if not df.index.is_monotonic_increasing: df df.sort_index()用min_periods代替window做兜底rolling(window3, min_periods2)表示只要凑够2个有效值就计算避免全NaN。生产环境里我们永远用min_periodsround(window*0.7)比如window7就设min_periods5确保数据轻微缺失时不中断。3.4 扩展窗口cumsum()不是“累加”而是“从起点到当前点的全量聚合”expanding().sum()看起来像cumsum()但二者有本质区别。cumsum()是纯粹的数学累加而expanding()是窗口函数支持任意聚合方法。比如expanding().std()计算的是“从第一条到当前条的标准差”这对监控非常关键——当某天标准差突然飙升说明客户行为基线正在漂移。但陷阱在于expanding()默认从第1行开始计算而第1行的std()是未定义的单个值无标准差会返回NaN。很多脚本直接忽略导致首日指标为空。正确做法是用min_periods2强制从第2行起算# 错误首日NaN且无法解释 df[cum_std] df[amount].expanding().std() # 正确首日仍为NaN但第2行起有值且逻辑清晰 df[cum_std] df[amount].expanding(min_periods2).std()更进一步银行合规要求所有累计指标必须标注“计算起始日”。我们在生产脚本里强制添加元数据df.attrs[expanding_start_date] df.index.min().strftime(%Y-%m-%d)这样当审计人员查数据时一眼就能看到“该累计值从2024-01-01开始计算”避免了无数扯皮。3.5 多级分组与unstack为什么unstack()后必须加fill_value0unstack()把行索引的一层转为列但默认遇到缺失组合会填NaN。在银行场景里NaN和0有天壤之别NaN代表“数据缺失不可信”0代表“该组合真实发生0次交易”。如果直接导出含NaN的报表业务方会质疑“是不是ETL漏数了”而填0则明确传递“此处无交易”的业务事实。但填0也有风险如果原始数据里真有NaN值比如某笔交易金额为空unstack()后也会变成NaN此时填0就篡改了事实。所以必须分两步走# 第一步区分“缺失组合”和“真实NaN” # 先用fillna()把原始数据的空值标为特殊标记 df_clean df.fillna({amount: -999999}) # 用极值标记空值 # 第二步分组聚合时过滤掉标记值 result df_clean.groupby([region,product])[amount].agg( lambda x: x[x ! -999999].mean() # 只对有效值计算 ).unstack(fill_value0) # 缺失组合填0真实空值已在上步过滤这套组合拳确保了报表里每一个0都是业务意义上的“零交易”而非技术意义上的“数据丢失”。3.6 端到端实战银行信用卡分析流水线的七个生死关原文的End-to-End示例很好但生产环境要补足七个致命细节关卡1数据生成的“业务真实性”原文用np.random.uniform(20,500,60)生成金额但真实信用卡交易有强分布特征小额高频奶茶25元、中额常态超市328元、大额低频机票4800元。我们用分位数采样法模拟# 基于真实交易分布20%小额(100), 65%中额(100-1000), 15%大额(1000) np.random.seed(42) amounts np.concatenate([ np.random.uniform(20, 99, sizeint(60*0.2)), np.random.uniform(100, 999, sizeint(60*0.65)), np.random.uniform(1000, 5000, sizeint(60*0.15)) ])关卡2时间排序的“绝对强制”原文df_sorted df_transactions.sort_values(date).set_index(date)但没处理重复日期。真实交易中同一秒可能有上千笔并发。必须加sort_index()二次保险df_sorted (df_transactions .sort_values([date,customer_id]) # 先按日期再按客户ID稳定排序 .set_index(date) .sort_index()) # 再按索引排序确保单调关卡3滚动窗口的“业务对齐”原文rolling(window7)是自然日但银行风控看的是交易日历剔除节假日。我们维护一张trading_calendar.csv用pd.offsets.CustomBusinessDay替换from pandas.tseries.offsets import CustomBusinessDay calendar pd.read_csv(trading_calendar.csv, parse_dates[date]) cbd CustomBusinessDay(calendarcalendar[date].dt.date.unique()) df_sorted[rolling_7day_avg] df_sorted.groupby(customer_id)[amount].rolling( window7, min_periods5, closedboth ).mean().reset_index(level0, dropTrue)关卡4自定义函数的“空值免疫”原文risk_metrics函数没处理series全为NaN的情况。生产版必须加def risk_metrics(series, high_value_threshold300): if series.isna().all(): return pd.Series({high_value_count: 0, high_value_pct: 0.0, regular_avg: 0.0}) # 后续逻辑...关卡5列名扁平化的“业务语义”原文summary.columns [total_spend,avg_transaction,...]是硬编码。我们用配置驱动AGG_CONFIG { amount: {sum: total_spend, mean: avg_transaction, count: txn_count}, fee: {sum: total_fees} } # 动态生成列名映射 new_cols [] for col, funcs in AGG_CONFIG.items(): for func, alias in funcs.items(): new_cols.append((col, func, alias)) # 扁平化时按配置映射关卡6内存优化的“分块聚合”当数据超亿行groupby().agg()会OOM。我们用chunksize分批def chunked_agg(file_path, chunk_size100000): results [] for chunk in pd.read_csv(file_path, chunksizechunk_size): chunk_result chunk.groupby([customer_id,category]).agg({...}) results.append(chunk_result) return pd.concat(results).groupby(level[0,1]).sum() # 二次聚合关卡7结果验证的“黄金快照”每次脚本更新必须和旧版结果比对。我们用deepdiff库做逐字段校验from deepdiff import DeepDiff old_result pd.read_parquet(gold_snapshot_v1.parquet) new_result run_new_pipeline() diff DeepDiff(old_result, new_result, ignore_orderTrue) if diff: raise RuntimeError(f聚合逻辑变更导致结果差异: {diff})4. 实操过程详解从原始交易表到高管仪表盘的完整链路4.1 数据准备银行级交易表的“脏数据预处理”真实银行的交易表远比示例复杂。我们以某股份制银行的card_transaction_raw表为例它有67列日增量2300万行。聚合前必须过三关第一关字段清洗Field Sanitizationtransaction_amount: 存在NULL字符串、-占位符、0.000实际为0但精度异常merchant_category: 有RETAIL 尾部空格、Dining 大小写混用、E-COMMERCE需映射为ECOMMERCEdate: 部分记录为20240101YYYYMMDD格式字符串标准化脚本def clean_transaction_data(df): # 金额列转数值异常值设为NaN df[transaction_amount] pd.to_numeric( df[transaction_amount].replace([NULL, -, ], np.nan), errorscoerce ).round(2) # 商户类型去空格、统一大小写、映射别名 category_map {E-COMMERCE: ECOMMERCE, ONLINE: ECOMMERCE} df[merchant_category] (df[merchant_category] .str.strip() .str.upper() .map(lambda x: category_map.get(x, x))) # 日期列强制转datetime无效值设为NaT df[date] pd.to_datetime(df[date], formatmixed, errorscoerce) return df # 应用清洗 df_clean clean_transaction_data(df_raw) print(f清洗后数据量: {len(df_clean)}, 金额空值率: {df_clean[transaction_amount].isna().mean():.2%})第二关业务过滤Business Filtering银行只分析“已清算”交易排除status ! SETTLED未清算、撤单、失败transaction_type REFUND退款需单独建模amount 0理论上不应存在但上游系统偶发错误df_filtered df_clean[ (df_clean[status] SETTLED) (df_clean[transaction_type] ! REFUND) (df_clean[transaction_amount] 0) ].copy()第三关衍生字段Feature Engineeringhour_of_day: 从date提取小时用于分时段分析is_weekend: 判断是否周末风控策略不同amount_tier: 按监管要求分档100,100-1000,1000df_filtered[hour_of_day] df_filtered[date].dt.hour df_filtered[is_weekend] df_filtered[date].dt.dayofweek 5 df_filtered[amount_tier] pd.cut( df_filtered[transaction_amount], bins[0, 100, 1000, float(inf)], labels[SMALL, MEDIUM, LARGE] )至此我们得到df_ready这才是聚合的真正起点。跳过这三步直接groupby就像没洗菜就下锅——味道全毁。4.2 多维聚合实施七种业务场景的代码实现现在我们基于df_ready实现银行真实的七类聚合需求。每段代码都附带业务背景、技术要点、避坑提示。场景1分行行长看板——按城市商户类型统计基础多维# 业务背景华东分行王行长要对比上海、杭州、南京三市的餐饮、零售商户表现 # 技术要点用query()预过滤避免全量分组unstack()后fill_value0 result_city_merchant ( df_ready.query(city in [Shanghai,Hangzhou,Nanjing] and merchant_category in [DINING,RETAIL]) .groupby([city,merchant_category]) .agg({ transaction_amount: [(avg_amt,mean), (med_amt,median), (std_amt,std)], transaction_count: [(cnt,sum)] }) .unstack(fill_value0) ) # 避坑提示std()对单值返回NaN需用min_periods2 result_city_merchant.columns [_.join(col).strip() for col in result_city_merchant.columns.values]场景2风控模型输入——滚动30日欺诈特征时间窗口# 业务背景反欺诈模型需要每个客户最近30天的交易频次波动率 # 技术要点用resample()替代rolling()按日重采样更稳定波动率用变异系数(CV) df_daily df_ready.set_index(date).groupby(customer_id)[transaction_amount].resample(D).count().reset_index(namedaily_txn_cnt) # 计算滚动30日CVstd/mean规避量纲影响 df_daily[cv_30d] ( df_daily.groupby(customer_id)[daily_txn_cnt] .rolling(window30, min_periods15) .apply(lambda x: x.std() / x.mean() if x.mean() ! 0 else 0, rawTrue) ) # 避坑提示rawTrue提升性能CV分母为0时返回0避免Inf传播场景3运营活动效果——同期群分析Cohort Analysis# 业务背景评估“新客首刷返现”活动看首刷当月vs次月的留存交易 # 技术要点用first()找首刷日计算月份差避免用date.month直接减跨年错误 df_cohort df_ready.copy() df_cohort[first_txn_date] df_cohort.groupby(customer_id)[date].transform(min) df_cohort[cohort_month] df_cohort[first_txn_date].dt.to_period(M) df_cohort[txn_month] df_cohort[date].dt.to_period(M) df_cohort[month_diff] (df_cohort[txn_month] - df_cohort[cohort_month]).apply(lambda x: x.n) # 聚合各同期群在各月的平均交易额 cohort_result ( df_cohort.groupby([cohort_month,month_diff]) .agg({transaction_amount: mean}) .unstack(fill_value0) )场景4监管报送——按客户风险等级分层业务逻辑嵌入# 业务背景向银保监报送“高风险客户交易集中度”定义近3月交易额TOP10%且单笔5000元占比30% def risk_concentration(series): # series是单个客户的交易额序列 if len(series) 10: # 样本太少不参与评级 return 0.0 top10_threshold np.percentile(series, 90) high_value_ratio (series 5000).sum() / len(series) return 1.0 if (series.max() top10_threshold) and (high_value_ratio 0.3) else 0.0 # 应用先按客户聚合再应用自定义函数 df_customer df_ready.groupby(customer_id)[transaction_amount].apply(risk_concentration).reset_index(nameis_high_risk) # 再按地区统计高风险客户占比 result_regulatory df_customer.merge(df_ready[[customer_id,region]], oncustomer_id).groupby(region)[is_high_risk].mean()场景5BI自助分析——动态维度切换灵活聚合# 业务背景BI平台允许用户拖拽任意维度组合需通用聚合函数 def dynamic_aggregation(df, group_cols, agg_config): :param group_cols: list, 如[region,merchant_category] :param agg_config: dict, 如{transaction_amount:[mean,sum], fee:[mean]} # 安全校验 missing_cols set(group_cols list(agg_config.keys())) - set(df.columns) if missing_cols: raise ValueError(f缺失字段: {missing_cols}) # 构建agg字典自动添加别名 agg_dict {} for col, funcs in agg_config.items(): for func in funcs: alias f{col}_{func} agg_dict[col] agg_dict.get(col, []) [(alias, func)] result df.groupby(group_cols).agg(agg_dict) result.columns [_.join(col) for col in result.columns] return result # 使用示例 result dynamic_aggregation( df_ready, group_cols[city,amount_tier], agg_config{transaction_amount:[mean,std], transaction_count:[sum]} )场景6实时大屏——增量聚合Streaming Aggregation# 业务背景总行大屏需实时显示“当前小时各商户类型交易额”延迟30秒 # 技术要点用stateful processing维护滚动状态 class HourlyAggregator: def __init__(self): self.hourly_state {} # {hour_key: {merchant: amount_sum}} def update(self, row): hour_key row[date].floor(H).strftime(%Y-%m-%d %H:00:00) merchant row[merchant_category] amount row[transaction_amount] if hour_key not in self.hourly_state: self.hourly_state[hour_key] {} self.hourly_state[hour_key][merchant] ( self.hourly_state[hour_key].get(merchant, 0) amount ) def get_current(self): # 返回最新一小时数据 latest_hour max(self.hourly_state.keys()) return pd.DataFrame(list(self.hourly_state[latest_hour].items()), columns[merchant_category,hourly_amount]) # 生产中此对象部署在Flink或Spark Structured Streaming中场景7模型训练数据——负采样平衡ML Ready Aggregation# 业务背景训练欺诈检测模型需对正常交易做负采样使正负样本1:1 # 技术要点先分组统计正样本数再按比例采样负样本 fraud_labels df_ready[is_fraud] # 0 or 1 positive_count fraud_labels.sum() negative_sample_size int(positive_count * 1.5) # 负样本1.5倍 # 分层负采样按商户类型保持分布一致 df_negative df_ready[fraud_labels 0] sampled_negative df_negative.groupby(merchant_category, group_keysFalse).apply( lambda x: x.sample(nmin(len(x), int(negative_sample_size * len(x) / len(df_negative))), random_state42) ) # 合并正负样本供模型训练 df_ml_ready pd.concat([ df_ready[fraud_labels 1], sampled_negative ]).sort_values(date)4.3 性能调优从“能跑”到“飞快”的五个关键动作即使代码逻辑完美数据量上来后仍可能慢如蜗牛。以下是我在银行生产环境验证有效的五大调优动作动作1用categorical类型替代object商户类型、地区等枚举字段用astype(category)可降内存70%提速3倍df_ready[merchant_category] df_ready[merchant_category].astype(category) df_ready[region] df_ready[region].astype(category)动作2预聚合Pre-aggregation对高频查询维度提前计算宽表# 每日执行一次生成customer_daily_summary customer_daily df_ready.groupby([customer_id,date]).agg({ transaction_amount: [sum,count,mean], fee: sum }).round(2) customer_daily.columns [daily_spend,daily_txn_cnt,daily_avg_amt,daily_fee] customer_daily.to_parquet(customer_daily_summary.parquet, indexTrue)后续分析直接读取此宽表避免每次重算。动作3分区读取Partitioned Reading用pyarrow读Parquet分区表只加载必要分区# 按日期分区存储 df_partitioned pd.read_parquet( s3://bucket/transactions/, filters[(date, , 2024-01-01), (date, , 2024-01-31)] )动作4Dask分布式聚合单机扛不住时无缝切到Daskimport dask.dataframe as dd df_dask dd.read_parquet(s3://bucket/transactions/) result (df_dask.groupby([region,merchant_category]) .agg({transaction_amount: [mean,std]})