1. 项目概述为什么“结构”比“功能”更早决定一个数据处理脚本的生死你写过这样的代码吗——一个.py文件开头import二十行中间三百行嵌套for循环加if判断最后十几行用pandas.read_csv硬生生把原始日志、Excel报表、JSON接口响应全塞进同一个DataFrame里再靠print(df.head())和反复注释/反注释来调试改需求时不敢动第87行因为怕影响第213行的fillna逻辑交接给同事时对方盯着屏幕沉默三分钟然后默默新建了一个notebook重写。这不是段子这是我过去三年在五家不同公司做数据工程支持时看到的最普遍、最隐蔽、也最消耗团队精力的“慢性病”。“How to Efficiently Structure Your Data Processing Code”这个标题表面看是讲代码组织技巧实则直指数据工作流中最常被忽视的底层基建问题结构即契约结构即文档结构即可维护性本身。它不解决“怎么算出结果”而是决定“这个结果能不能被下个月的你、隔壁组的新人、甚至审计系统稳定复现”。我见过太多项目算法模型准确率98%但因数据清洗逻辑散落在三个不同脚本里、命名全是tmp_v2_final_corrected.csv最终上线后因上游字段微调导致整条链路静默失败三天才被发现。这篇文章不是教你怎么用pandas高级函数而是带你从第一行代码开始就建立一套能扛住需求变更、人员流动、数据源漂移的防御性结构体系。无论你是刚学完Python基础的数据分析新手还是带团队做ETL架构的资深工程师只要你的工作涉及“把原始数据变成可用信息”这篇就是为你写的实战手册。2. 整体设计思路三层隔离原则与“数据契约”的强制落地2.1 为什么必须放弃“单文件万能脚本”思维很多人觉得“结构”是大厂或复杂系统的专利小项目写个脚本够用就行。这种想法错在混淆了“功能实现”和“系统韧性”。我试过用单文件处理一个电商订单清洗任务原始数据来自MySQL导出CSV、微信小程序埋点JSON、客服工单Excel目标是生成每日销售漏斗报表。初期确实快——50行代码搞定。但当业务方提出“把退款订单单独标记为灰色并在漏斗第二步排除”时问题来了退款标识在JSON里叫is_refunded在Excel里是refund_status列含文字“已退款”在MySQL里却是order_status4。我不得不在原脚本里加三处if判断又因JSON解析失败导致整个流程中断只好临时加try-except结果错误日志里全是“KeyError: is_refunded”却找不到具体哪条记录出错。这就是单文件结构的致命伤所有关注点数据源接入、业务规则、输出格式全部耦合一处修改引发全局震荡且无明确边界定位问题。后来我按三层结构重写耗时2小时但后续6次需求变更平均每次只改1个文件、不超过10行代码且每次修改后都能独立测试零线上事故。2.2 三层隔离输入层、处理层、输出层的物理切割我坚持的结构核心是物理隔离契约驱动而非逻辑分组。所谓“物理隔离”是指三个层级必须对应三个独立文件或模块且禁止跨层直接调用。具体如下输入层ingest/只做一件事——把原始数据“原样搬进来”不做任何清洗、转换、过滤。例如ingest/mysql_orders.py只负责连接数据库、执行SELECT * FROM orders返回一个标准pandas.DataFrame列名严格匹配原始表ingest/wechat_logs.py只解析JSON数组对缺失字段统一填None绝不尝试补全或推断。这一层的契约是“输入模块输出的DataFrame其列名、数据类型、空值定义必须与原始数据源完全一致”。处理层transform/这是唯一允许业务逻辑存在的地方。它接收输入层输出的原始DataFrame通过一系列纯函数无副作用进行转换。例如transform/sales_funnel.py包含clean_order_status(df)、flag_refunds(df)、calculate_conversion_rate(df)等函数每个函数只接收df只返回新df不修改原df。关键点在于所有函数必须有明确的输入/输出契约声明我用type hints强制约束from typing import List, Dict, Optional import pandas as pd def flag_refunds(raw_df: pd.DataFrame) - pd.DataFrame: 输入原始订单DataFrame必须包含列[order_id, is_refunded, refund_status, order_status] 输出新增列refund_flagboolTrue表示应计入退款 契约若原始数据中refund_flag列已存在则抛出ValueError if refund_flag in raw_df.columns: raise ValueError(Input DataFrame already contains refund_flag column) # 实际逻辑... return processed_df这种契约让每个函数成为可插拔的积木测试时只需mock输入df验证输出df是否符合预期。输出层export/只负责格式化和落库。export/daily_report.py接收处理层输出的标准化DataFrame调用to_csv()、to_sql()或API推送但绝不碰业务规则。它的契约是“输出模块只接受已通过处理层校验的DataFrame且仅修改数据格式如日期转字符串、数值四舍五入不改变业务语义”。提示三层之间通过明确定义的DataFrame Schema进行通信而非隐式约定。我用pandera库在每层入口添加schema校验例如输入层返回前执行schema.validate(raw_df)处理层函数入口检查input_schema.validate(df)输出层入口检查output_schema.validate(df)。这相当于给数据流装上“安检仪”问题在源头就被拦截而不是等到报表生成失败才去回溯。2.3 “数据契约”的四个强制维度光分层不够必须用契约固化各层责任。我定义的契约包含四个不可妥协的维度列名契约Column Name Contract所有输入层输出的DataFrame列名必须与原始数据源100%一致包括大小写、下划线。处理层函数若需新增列必须在文档中明确声明且列名遵循snake_case规范如user_lifetime_value禁用驼峰或中文。曾有个项目因MySQL表字段create_time被输入层自动转成CreateTime导致处理层filter_by_date(df)函数里写的df[df[create_time] threshold]永远返回空——因为列名根本不存在。从此我强制所有输入模块在返回前执行df.columns df.columns.str.lower().str.replace( , _)。数据类型契约Data Type Contract输入层必须将原始数据映射为pandas标准类型int64, float64, string, datetime64[ns], boolean禁用object类型存储数字或日期。例如Excel里的“2023-01-01”字符串输入层必须用pd.to_datetime()转为datetime64而不是留着string。处理层函数的type hints必须精确到具体类型def calculate_age(birth_date: pd.Series) - pd.Series:而非def calculate_age(birth_date: pd.Series) - Any:。空值契约Null Value Contract明确规定每列空值的语义。例如order_amount列的NaN表示“金额未录入”而0表示“免费订单”user_phone列的NaN表示“用户未提供”空字符串表示“用户明确填写了空值”。处理层所有函数必须基于此语义做逻辑如fill_missing_phone(df)只能用指定默认值填充NaN绝不能误填空字符串。行级契约Row-Level Contract对数据行的唯一性、完整性做约束。例如输入层ingest/user_profiles.py必须保证返回的DataFrame中user_id列无重复通过df.drop_duplicates(subset[user_id])并记录去重数否则抛出IntegrityError。处理层deduplicate_orders(df)函数必须声明“输入df中order_id必须唯一否则输出结果不可预测”。这套契约不是纸上谈兵。我在一个金融风控项目中用pandera定义了包含23个字段的schema每次数据进入处理层前自动校验两周内捕获了7次上游数据源变更如新增risk_score_v2列、credit_limit类型从int变float全部在开发环境预警避免了生产事故。契约的本质是把“人脑记忆的隐式规则”变成“机器可执行的显式约束”。3. 核心细节解析从目录结构到函数设计的12个实操要点3.1 目录结构为什么src/下必须有ingest/、transform/、export/三个平行目录很多团队会把代码放在scripts/或notebooks/下认为“结构是后期优化的事”。这是巨大误区。目录结构是第一个契约载体它向所有协作者宣告“这里只放输入逻辑”、“这里只放转换逻辑”。我的标准结构长这样project_root/ ├── src/ │ ├── ingest/ # 输入层只读不写 │ │ ├── __init__.py │ │ ├── mysql_orders.py │ │ ├── wechat_logs.py │ │ └── s3_raw_data.py │ ├── transform/ # 处理层纯函数无副作用 │ │ ├── __init__.py │ │ ├── sales_funnel.py │ │ ├── user_segmentation.py │ │ └── fraud_detection.py │ ├── export/ # 输出层只写不读业务逻辑 │ │ ├── __init__.py │ │ ├── daily_report.py │ │ ├── dashboard_api.py │ │ └── audit_log.py │ └── utils/ # 工具层跨层复用但不得包含业务规则 │ ├── __init__.py │ ├── logging_config.py │ └── schema_validator.py ├── tests/ # 测试必须1:1对应src结构 │ ├── test_ingest/ │ │ └── test_mysql_orders.py │ ├── test_transform/ │ │ └── test_sales_funnel.py │ └── test_export/ │ └── test_daily_report.py ├── config/ # 配置中心化禁止硬编码 │ ├── database.yaml │ ├── api_endpoints.yaml │ └── schemas/ # 所有pandera schema定义 │ ├── input_schemas.py │ └── output_schemas.py └── main.py # 唯一入口串联三层无业务逻辑关键细节在于__init__.py的用法。每个子目录的__init__.py只做两件事1暴露该层的核心函数2设置该层的默认配置。例如ingest/__init__.py# src/ingest/__init__.py from .mysql_orders import fetch_orders from .wechat_logs import parse_logs from .s3_raw_data import load_from_s3 # 暴露统一入口隐藏具体实现 def load_all_sources() - dict: 返回字典{orders: df, logs: df, raw_data: df} return { orders: fetch_orders(), logs: parse_logs(), raw_data: load_from_s3() } # 设置该层默认超时、重试策略 DEFAULT_TIMEOUT 30 MAX_RETRIES 3这样处理层transform/sales_funnel.py只需from src.ingest import load_all_sources无需关心数据从哪来、怎么连。当某天微信日志改用Kafka接入时我只改ingest/wechat_logs.py和ingest/__init__.py里的parse_logs实现其他所有代码零修改。目录结构的价值在于让“变化”被限制在最小物理范围内。3.2 输入层实操如何应对“原始数据源永不按约定出牌”的现实理想很丰满现实很骨感。上游数据源永远在变MySQL表突然加字段、Excel模板换颜色、JSON接口返回null代替空数组。输入层的使命不是“完美适配”而是“可控降级”。我的输入层函数设计遵循三个铁律Fail Fast, Fail Loud遇到无法解析的数据立即抛出带上下文的异常而非默默跳过或填默认值。例如解析JSON日志时def parse_logs(json_path: str) - pd.DataFrame: with open(json_path, r) as f: try: data json.load(f) except json.JSONDecodeError as e: raise ValueError(fInvalid JSON in {json_path}: {e.msg} at line {e.lineno}) if not isinstance(data, list): raise ValueError(fExpected JSON array in {json_path}, got {type(data).__name__}) # 关键记录原始数据样本用于debug sample_record data[0] if data else {} if event_type not in sample_record: raise KeyError(fMissing required field event_type in first record of {json_path}. Sample: {sample_record})这样当上游把event_type改成eventType时错误信息直接告诉你“第一行记录缺少event_type样本是{...}”而不是在下游报KeyError: event_type让你猜是哪条记录。Schema柔化Schema Softening不强求输入数据100%符合预期schema而是定义“必须字段”和“可选字段”。用pandera的coerceTrue参数自动类型转换并用nullableTrue容忍空值import pandera as pa from pandera.typing import Series class RawLogSchema(pa.SchemaModel): event_type: Series[str] pa.Field(coerceTrue, nullableFalse) timestamp: Series[pa.DateTime] pa.Field(coerceTrue, nullableTrue) # 可选缺失则填NaT user_id: Series[str] pa.Field(coerceTrue, nullableTrue) # 可选缺失则填None # 新增字段自动忽略不报错 class Config: coerce True # 自动类型转换 strict False # 允许额外列版本化快照Versioned Snapshot为高风险数据源如第三方API保存原始数据快照。ingest/s3_raw_data.py在加载S3文件后自动生成带时间戳的备份def load_from_s3(bucket: str, key: str) - pd.DataFrame: # ... 加载逻辑 # 保存快照用于审计和回滚 snapshot_key fsnapshots/{key.replace(/, _)}_{int(time.time())}.parquet df.to_parquet(fs3://{bucket}/{snapshot_key}) return df当某天API返回异常数据导致下游崩溃我能立刻从S3找到2小时前的快照用load_from_s3(..., snapshot_key...)恢复而不是干等上游修复。注意输入层绝对禁止做业务逻辑。曾有个同事在ingest/mysql_orders.py里写了if order_status cancelled: continue理由是“脏数据太多先过滤掉”。结果三个月后业务方说“取消订单要计入流失分析”我们才发现这部分数据从未进入处理链路。输入层的黄金法则是“原样搬运不增不减不评不判”。3.3 处理层函数设计为什么“纯函数”和“链式调用”是可维护性的基石处理层是业务逻辑的心脏也是最容易腐化的部分。我坚持两个设计原则所有函数必须是纯函数Pure Function所有流程必须是链式调用Chaining。纯函数意味着1相同输入永远产生相同输出2不修改外部状态不改全局变量、不写文件、不调API3不依赖外部时间如datetime.now()、随机数等。例如计算用户生命周期价值LTV# ✅ 正确纯函数输入输出明确 def calculate_user_ltv( orders_df: pd.DataFrame, users_df: pd.DataFrame, discount_rate: float 0.1, max_periods: int 12 ) - pd.DataFrame: # 所有计算基于输入df不读config、不查DB ltv_df orders_df.merge(users_df, onuser_id) ltv_df[ltv] ltv_df[order_amount] / ((1 discount_rate) ** ltv_df[period]) return ltv_df[[user_id, ltv]].groupby(user_id).sum().reset_index() # ❌ 错误依赖外部状态不可测试 def calculate_user_ltv_bad(): # 读取全局配置 config get_config() # 依赖外部 # 查询数据库获取最新discount_rate rate db.query(SELECT rate FROM discount_table WHERE active1) # 依赖IO # 使用当前时间 now datetime.now() # 依赖时间链式调用意味着业务流程由多个纯函数按顺序组合而成形成清晰的数据流。我用functools.reduce或自定义pipe函数实现from functools import reduce def pipe(data, *functions): 将data依次传入functions返回最终结果 return reduce(lambda acc, fn: fn(acc), functions, data) # 在main.py中定义业务流程 def build_sales_funnel(): raw_data ingest.load_all_sources() funnel_df pipe( raw_data[orders], transform.clean_order_status, transform.flag_refunds, transform.aggregate_by_stage, transform.calculate_conversion_rates ) export.save_daily_report(funnel_df)这样做的好处是爆炸性的1每个函数可独立单元测试2调试时可在任意环节print()中间结果3A/B测试时只需替换一个函数如transform.flag_refunds_v24性能分析时用%timeit逐个测量函数耗时。实操心得处理层函数命名必须体现“动作对象”如clean_order_status、enrich_user_profile、validate_transaction_flow禁用模糊词如process_data、handle_stuff。我曾重构一个命名do_magic()的函数发现它实际做了5件事去重、填空、转类型、过滤、排序。拆分成5个函数后测试覆盖率从32%升到91%且其中3个函数被其他项目复用。3.4 输出层陷阱为什么“格式化”比“计算”更容易引发线上事故输出层看似简单却是线上事故高发区。原因在于它直连外部系统数据库、API、文件系统且错误往往不报错只“静默失败”。我总结了输出层四大雷区及解法雷区典型表现解决方案实操案例类型失真Type Mismatchpandas的int64写入MySQL的TINYINT超范围变NULL输出前强制类型转换用astype()明确指定目标类型df[status_code] df[status_code].astype(int8)并在写入前用df.dtypes校验时区混乱Timezone Confusion本地时间写入UTC数据库报表显示时间错6小时所有datetime列在输出层统一转为UTC用dt.tz_localize(None).dt.tz_localize(UTC)df[created_at] df[created_at].dt.tz_localize(Asia/Shanghai).dt.tz_convert(UTC)主键冲突Primary Key Conflict同一订单ID多次写入数据库报Duplicate Entry输出前检查目标表是否存在用ON CONFLICT DO UPDATEPostgreSQL或REPLACE INTOMySQLPostgreSQL中用sqlalchemy.dialects.postgresql.insert().on_conflict_do_update()大文件阻塞Large File Blocking导出10GB CSV卡死进程占用全部内存分块写入用chunksize参数控制批次df.to_sql(report_table, conengine, chunksize10000, if_existsreplace)最关键的是输出层必须自带健康检查。每个输出函数执行后必须验证结果是否符合预期。例如export.save_daily_report(df)def save_daily_report(df: pd.DataFrame, target_table: str daily_funnel): # 写入数据库 df.to_sql(target_table, conengine, if_existsreplace, indexFalse) # 健康检查验证行数、关键指标范围 result_count engine.execute(fSELECT COUNT(*) FROM {target_table}).scalar() if result_count ! len(df): raise RuntimeError(fRow count mismatch: expected {len(df)}, got {result_count}) # 验证转化率在合理范围0-100% conversion_rate engine.execute(fSELECT AVG(conversion_rate) FROM {target_table}).scalar() if not (0 conversion_rate 100): raise ValueError(fInvalid conversion_rate: {conversion_rate}%) logger.info(fDaily report saved to {target_table}, {result_count} rows)这个检查花了我30秒写却在一次上游数据异常时提前2小时发现了转化率突变为1200%的问题避免了错误报表被发送给CEO。4. 实操过程从零搭建一个电商销售漏斗处理流水线4.1 环境准备与依赖管理为什么pyproject.toml比requirements.txt更适合数据项目很多数据项目还用pip install -r requirements.txt这在协作中埋下巨坑。requirements.txt只记录最终版本不区分直接依赖和传递依赖升级时容易破坏。我全面转向pyproject.tomlPEP 518标准用poetry管理依赖。核心优势在于精确锁定Exact Lockingpoetry.lock文件记录每个包的完整哈希值确保poetry install在任何机器上安装完全相同的依赖树。环境隔离Environment Isolationpoetry shell自动激活虚拟环境避免pip install污染系统Python。分组依赖Group Dependencies明确区分生产依赖、开发依赖、测试依赖。我的pyproject.toml精简版[tool.poetry] name ecommerce-funnel version 0.1.0 description Sales funnel processing pipeline [tool.poetry.dependencies] python ^3.9 pandas ^2.0.0 numpy ^1.24.0 pandera ^0.17.0 sqlalchemy ^2.0.0 psycopg2-binary ^2.9.0 [tool.poetry.group.dev.dependencies] pytest ^7.0.0 pytest-cov ^4.0.0 black ^23.0.0 jupyter ^1.0.0 [tool.poetry.group.test.dependencies] pytest-mock ^3.10.0 responses ^0.23.0 [build-system] requires [poetry-core] build-backend poetry.core.masonry.api安装步骤极简# 1. 安装poetry一次 curl -sSL https://install.python-poetry.org | python3 - # 2. 进入项目目录安装所有依赖含dev组 poetry install # 3. 启动shell自动激活虚拟环境 poetry shell # 4. 运行测试 poetry run pytest tests/注意poetry会自动创建.venv目录但绝不提交到Git。我在.gitignore中明确添加/.venv、/poetry.lock等等lock文件必须提交。poetry.lock是保证可重现性的核心必须和代码一起提交。曾有个项目因没提交lock文件CI服务器安装了新版pandas导致df.groupby().apply()行为变更漏斗计算结果偏差15%排查了两天才发现是依赖问题。4.2 第一步构建输入层——从三个异构源加载原始数据我们模拟一个真实场景电商销售漏斗需要三份原始数据——MySQL订单表、微信小程序用户行为JSON、客服工单Excel。按三层结构先建src/ingest/。1. MySQL订单接入 (src/ingest/mysql_orders.py)使用SQLAlchemy连接关键点是只读、只选、不加工from sqlalchemy import create_engine, text import pandas as pd from src.utils.logging_config import get_logger from src.config.database import MYSQL_CONFIG logger get_logger(__name__) def fetch_orders() - pd.DataFrame: 从MySQL读取原始订单数据列名与表结构100%一致 engine create_engine( fmysqlpymysql://{MYSQL_CONFIG[user]}:{MYSQL_CONFIG[password]}{MYSQL_CONFIG[host]}:{MYSQL_CONFIG[port]}/{MYSQL_CONFIG[database]} ) # 关键SELECT *不加WHERE不JOIN不聚合 query text(SELECT * FROM orders) try: df pd.read_sql(query, engine) logger.info(fFetched {len(df)} orders from MySQL) return df except Exception as e: logger.error(fMySQL fetch failed: {e}) raise2. 微信JSON日志接入 (src/ingest/wechat_logs.py)处理JSON数组重点是柔化schema、记录样本import json import pandas as pd from src.utils.logging_config import get_logger from pandera import DataFrameSchema, Column, Check from pandera.typing import Series logger get_logger(__name__) # 定义柔化schema WECHAT_LOG_SCHEMA DataFrameSchema({ event_type: Column(str, coerceTrue), timestamp: Column(datetime64[ns], coerceTrue, nullableTrue), user_id: Column(str, coerceTrue, nullableTrue), page_url: Column(str, coerceTrue, nullableTrue), duration_ms: Column(int, coerceTrue, nullableTrue), }, strictFalse, coerceTrue) # 允许额外列自动类型转换 def parse_logs(json_path: str) - pd.DataFrame: 解析微信日志JSON返回标准化DataFrame with open(json_path, r) as f: try: data json.load(f) except json.JSONDecodeError as e: raise ValueError(fInvalid JSON in {json_path}: {e.msg}) if not isinstance(data, list): raise ValueError(fExpected JSON array, got {type(data)}) # 记录样本用于debug if data: logger.debug(fJSON sample: {data[0]}) df pd.json_normalize(data) # 自动展开嵌套JSON # 应用schema柔化 try: validated_df WECHAT_LOG_SCHEMA.validate(df) logger.info(fParsed {len(validated_df)} logs from {json_path}) return validated_df except Exception as e: logger.error(fSchema validation failed for {json_path}: {e}) raise3. Excel工单接入 (src/ingest/support_tickets.py)处理Excel的坑最多合并单元格、多表头、空行关键是预处理显式列名import pandas as pd from src.utils.logging_config import get_logger logger get_logger(__name__) def load_tickets(excel_path: str) - pd.DataFrame: 加载客服工单Excel处理常见脏数据 try: # 关键跳过前3行标题/说明指定表头行为第4行 df pd.read_excel(excel_path, skiprows3, header0) # 删除全空行 df df.dropna(howall) # 重命名列强制snake_case去除空格和特殊字符 df.columns ( df.columns .str.strip() .str.replace(r[^a-zA-Z0-9_], _, regexTrue) .str.replace(r_, _, regexTrue) .str.lower() ) # 确保关键列存在缺失则添加 required_cols [ticket_id, user_id, status, created_date] for col in required_cols: if col not in df.columns: df[col] None logger.info(fLoaded {len(df)} tickets from {excel_path}) return df except Exception as e: logger.error(fExcel load failed for {excel_path}: {e}) raise4. 输入层统一入口 (src/ingest/__init__.py)暴露简洁API隐藏实现细节from .mysql_orders import fetch_orders from .wechat_logs import parse_logs from .support_tickets import load_tickets def load_all_sources() - dict: 统一加载所有原始数据源返回字典 return { orders: fetch_orders(), logs: parse_logs(data/raw/wechat_logs.json), tickets: load_tickets(data/raw/support_tickets.xlsx) }4.3 第二步构建处理层——实现销售漏斗的核心转换逻辑销售漏斗核心是浏览商品 → 加入购物车 → 提交订单 → 支付成功 → 确认收货。我们需要将三个数据源关联计算各阶段转化率。1. 数据关联与清洗 (src/transform/funnel_join.py)纯函数输入原始数据字典输出关联后的宽表import pandas as pd from src.utils.logging_config import get_logger logger get_logger(__name__) def join_funnel_data(sources: dict) - pd.DataFrame: 将订单、日志、工单数据关联构建漏斗宽表 输入{orders: df, logs: df, tickets: df} 输出宽表包含所有漏斗阶段字段 orders sources[orders].copy() logs sources[logs].copy() tickets sources[tickets].copy() # 步骤1从日志提取浏览和加购行为 # 浏览event_type page_view page_views logs[logs[event_type] page_view].copy() page_views page_views.rename(columns{timestamp: view_time}) # 加购event_type add_to_cart add_to_carts logs[logs[event_type] add_to_cart].copy() add_to_carts add_to_carts.rename(columns{timestamp: cart_time}) # 步骤2关联用户行为用user_id funnel_df orders[[order_id, user_id, order_time, status]].copy() # 左连接浏览行为一个用户可能浏览多次取最早一次 if not page_views.empty: first_view page_views.groupby(user_id)[view_time].min().reset_index(namefirst_view_time) funnel_df funnel_df.merge(first_view, onuser_id, howleft) # 左连接加购行为 if not add_to_carts.empty: first_cart add_to_carts.groupby(user_id)[cart_time].min().reset_index(namefirst_cart_time) funnel_df funnel_df.merge(first_cart, onuser_id, howleft) # 步骤3标记各阶段完成情况 funnel_df[viewed] funnel_df[first_view_time].notna() funnel_df[carted] funnel_df[first_cart_time].notna() funnel_df[ordered] funnel_df[order_id].notna() funnel_df[paid] funnel_df[status].isin([paid, completed]) funnel_df[received] funnel_df[status] delivered logger.info(fJoined funnel data, {len(funnel_df)} rows) return funnel_df2. 计算转化率 (src/transform/funnel_metrics.py)纯函数输入宽表输出指标DataFrameimport pandas as pd from src.utils.logging_config import get_logger logger get_logger(__name__) def calculate_funnel_metrics(funnel_df: pd.DataFrame) - pd.DataFrame: 计算各阶段转化率 输入join_funnel_data输出的宽表 输出指标DataFrame列stage, count, conversion_rate stages [ (viewed, 浏览商品), (carted, 加入购物车), (ordered, 提交订单), (paid, 支付成功), (received, 确认收货) ] metrics [] total_users len(funnel_df) for col, name in stages: count funnel_df[col].sum() # 转化率 当前阶段人数 / 上一阶段人数首阶段除以总用户数 if col viewed: rate count / total_users if total_users 0 else 0 else: # 找上一阶段列名 prev_col stages[stages.index((col, name)) - 1][0] prev_count funnel_df[prev_col].sum() rate count / prev_count if prev_count 0 else 0 metrics.append({ stage: name, count: int(count), conversion_rate: round(rate * 100, 2) }) metrics_df pd.DataFrame(metrics) logger.info(Calculated funnel metrics) return metrics_df3. 处理层统一入口 (src/transform/__init__.py)from .funnel_join import join_funnel_data from .funnel_metrics import calculate_funnel_metrics def build_funnel_pipeline(sources: dict) - pd.DataFrame: 端到端漏斗处理管道 joined_df