企业官网建设流程全解析

1. 为什么“ETL”不是过时的老古董，而是每天都在 quietly run 的数据心脏

很多人第一次听说ETL，是在某次技术分享会上被带过一句：“哦，就是把数据从A库抽出来，洗一洗，再塞进B库。”然后就默默划走——觉得这不就是个搬运工活儿？配不上“架构”“中台”“实时数仓”这些闪亮新词。我刚入行那会儿也这么想，直到在某跨平台系统项目里连续三天守着凌晨两点的调度日志，看着一条订单清洗规则漏掉了小数点后两位精度，导致下游财务报表差了73.6万元，而问题根源竟是一段正则表达式里少写了?量词。那一刻我才真正明白：ETL不是管道，是数据世界的交通管制中心；它不炫技，但一旦出错，整个业务链路都会堵死。

ETL（Extract-Transform-Load）这三个字母背后，承载的是数据从混沌到有序、从不可信到可决策的完整信任构建过程。它不追求毫秒级响应，但要求100%的语义准确；它不强调算法复杂度，但极度依赖对业务逻辑的深度咬合。比如“用户注册时间”字段，在源系统里可能是字符串格式2024-03-15T09:22:17+08:00，在CRM里要转成标准日期类型，而在风控模型里却需要拆解为“注册小时段”“是否工作日”两个离散特征——同一份原始数据，在不同下游场景中，transform的路径完全不同。这不是简单的格式转换，而是业务语义的翻译工程。

你不需要立刻上手写Spark作业或部署Airflow集群。真正的ETL入门，是从理解“为什么必须分三步走”开始的。Extract不是复制粘贴，是建立数据契约；Transform不是写SQL函数，是定义业务规则；Load不是INSERT INTO，是保障一致性边界。本文不讲抽象理论，只带你用最朴素的Python+Pandas+SQLite组合，亲手跑通一个真实可验证的端到端流程：从模拟电商订单库抽取原始数据，清洗地址字段中的乱码与空格、标准化商品类目编码、补全缺失的用户等级标签，最后加载进分析型宽表，并验证每一步的数据血缘关系。所有代码可直接复制运行，所有陷阱我都替你踩过了。

提示：本文所有操作均基于本地环境完成，无需云服务、无需数据库权限申请、无需运维配合。你只需要一台装有Python 3.8+的电脑，和23分钟专注时间。

2. Extract阶段：不是“拿数据”，而是“签协议”——如何安全、可控、可追溯地获取源头数据

很多人把Extract简单理解为“连上数据库，SELECT * FROM table”。这是最大的认知偏差。Extract的本质，是建立数据提供方（Source）与数据消费方（Target）之间的第一份契约——它规定了“谁可以拿”“拿什么”“什么时候拿”“拿多少”“拿错了怎么办”。跳过契约设计，直接开干，后面90%的问题都源于此。

2.1 源头数据建模：用最小必要字段还原业务事实

我们以模拟电商订单场景为例。真实业务中，订单数据分散在多个表：orders（主单）、order_items（子项）、users（用户）、products（商品）。但ETL的第一步，绝不是一股脑全拉过来。我们必须先回答：本次ETL的目标是什么？——构建一份用于销售分析的宽表，包含订单ID、下单时间、用户ID、用户等级、商品类目、订单金额、收货省份。

因此，Extract阶段只应获取最小必要集合：

• orders.id,orders.created_at,orders.user_id,orders.total_amount
• users.level（用户等级）
• order_items.product_id,order_items.quantity
• products.category_code（商品类目编码）

注意：我们不取users.address_detail（详细地址）、products.description（商品描述）等非分析字段。它们不仅增大传输体积，更可能引入敏感信息泄露风险。ETL不是数据大挪移，而是精准狙击。

我用SQLite模拟源库，建表语句如下（实际项目中，这里对应的是生产MySQL/Oracle的只读账号连接）：

-- 模拟源库：source_db.db CREATE TABLE orders ( id TEXT PRIMARY KEY, created_at TEXT NOT NULL, -- ISO8601字符串 user_id INTEGER NOT NULL, total_amount REAL NOT NULL ); CREATE TABLE users ( id INTEGER PRIMARY KEY, level TEXT CHECK(level IN ('VIP', 'GOLD', 'SILVER', 'BRONZE')) DEFAULT 'BRONZE' ); CREATE TABLE order_items ( id INTEGER PRIMARY KEY, order_id TEXT NOT NULL, product_id INTEGER NOT NULL, quantity INTEGER NOT NULL ); CREATE TABLE products ( id INTEGER PRIMARY KEY, category_code TEXT NOT NULL );

关键细节来了：orders.created_at是TEXT类型，而非DATETIME。这是很多老系统的真实现状——时间字段存为字符串，格式不统一（有的带时区，有的不带）。Extract阶段必须明确记录这个“缺陷”，它将成为后续Transform的处理依据。我在extract_config.yaml中这样声明：

source: db_path: "source_db.db" tables: - name: "orders" columns: ["id", "created_at", "user_id", "total_amount"] # 显式标注字段异常 anomalies: created_at: "string_format_iso8601_with_timezone" - name: "users" columns: ["id", "level"]

这份配置文件，就是Extract阶段签署的契约。它不解决数据质量问题，但确保所有参与方对“数据长什么样”达成共识。

2.2 抽取策略选择：全量、增量还是变更捕获？选错等于白干

面对每日百万级订单，你不可能每次都全量SELECT。必须根据业务容忍度选择抽取策略：

本入门实践采用时间戳增量，因其平衡性最好。我们约定：每次抽取created_at >= 上次抽取最大时间的数据。但这里有个致命陷阱——created_at是字符串！直接用WHERE created_at > '2024-03-15'会触发字符串字典序比较，'2024-03-2' > '2024-03-15'结果为True（因为'2'>'1'），导致漏数据。

正确做法是：在Extract层就做一次轻量解析，将字符串转为可比较的时间对象，并存入临时DataFrame：

import pandas as pd from datetime import datetime def extract_orders(db_path: str, last_max_time: str = None) -> pd.DataFrame: conn = sqlite3.connect(db_path) # 先查出所有数据（小数据集演示用） df = pd.read_sql_query("SELECT id, created_at, user_id, total_amount FROM orders", conn) # 关键：立即解析时间，避免后续误用字符串 df['created_at_parsed'] = pd.to_datetime( df['created_at'], format='ISO8601', errors='coerce' # 解析失败设为NaT ) # 若指定last_max_time，则过滤 if last_max_time: cutoff = pd.to_datetime(last_max_time) df = df[df['created_at_parsed'] >= cutoff].copy() conn.close() return df

看到没？Extract阶段的“解析”动作，不是Transform的职责，而是Extract的自我保护。它把模糊的字符串，固化为明确的datetime类型，切断后续环节误用源头的风险。

2.3 元数据采集：没有血缘追踪的ETL，就像没有地图的远征

Extract阶段必须同步采集元数据（Metadata），否则当下游报表出错时，你根本不知道问题出在哪个环节。我们采集三类元数据：

1. 结构元数据：表名、字段名、字段类型、空值率
2. 处理元数据：抽取开始时间、结束时间、记录数、错误数
3. 业务元数据：字段业务含义、数据来源系统、负责人

用一个极简的extraction_log.json记录：

{ "run_id": "etl_20240315_0215", "start_time": "2024-03-15T02:15:22.123Z", "end_time": "2024-03-15T02:15:25.456Z", "tables": [ { "name": "orders", "record_count": 1247, "null_rate": {"created_at": 0.0, "user_id": 0.2}, "source_system": "legacy_order_core_v2" } ] }

这个日志文件，就是ETL的“黑匣子”。某天发现用户等级统计异常，你只需查run_id=etl_20240315_0215的日志，就能确认当时users.level字段空值率是20%，进而定位到是上游用户表同步任务故障——而不是在Transform代码里大海捞针。

注意：元数据采集必须原子化。如果Extract成功但日志写入失败，整个任务应视为失败。宁可重跑，不可留脏。

3. Transform阶段：清洗与转换不是“修修补补”，而是业务规则的代码化落地

如果说Extract是签合同，那么Transform就是履行合同。它把原始数据，按照业务契约，翻译成下游系统能理解、能信赖、能计算的形态。这里没有银弹，只有对业务细节的死磕。我见过太多团队把Transform写成“一堆if-else”，结果三个月后没人敢动——因为没人知道那个if status == 'P'里的P代表“已支付”还是“待支付”。

3.1 清洗（Cleaning）：直面数据世界的“脏、乱、差”

清洗不是美化数据，是修复数据与业务现实的断裂。我们聚焦三个高频痛点：

痛点1：地址字段的“幽灵空格”与乱码
源数据中users.address常含不可见字符（如\u200b零宽空格）、全角空格（ ）、多余换行。直接strip()无效。

实测方案：

import re def clean_address(address: str) -> str: if not isinstance(address, str): return "" # 1. 移除所有Unicode控制字符（含零宽空格） address = re.sub(r'[\u200b-\u200f\u202a-\u202e]', '', address) # 2. 替换全角空格为半角 address = address.replace(' ', ' ') # 3. 合并连续空格，首尾strip address = re.sub(r'\s+', ' ', address).strip() return address # 在Transform主流程中调用 df['cleaned_address'] = df['address'].apply(clean_address)

痛点2：商品类目编码的“方言”混乱
源系统中，手机类目可能写作"MOBILE","mobile","Mobile Phone","PHONE"。业务要求统一为"PHONE"。

错误做法：写4个replace()。
正确做法：建立映射词典+模糊匹配兜底：

CATEGORY_MAPPING = { "MOBILE": "PHONE", "mobile": "PHONE", "Mobile Phone": "PHONE", "PHONE": "PHONE", "TABLET": "TABLET", "LAPTOP": "LAPTOP" } def normalize_category(raw_cat: str) -> str: if not raw_cat: return "OTHER" # 精确匹配优先 if raw_cat in CATEGORY_MAPPING: return CATEGORY_MAPPING[raw_cat] # 小写后匹配 if raw_cat.lower() in [k.lower() for k in CATEGORY_MAPPING.keys()]: # 找到原key，返回对应value for k, v in CATEGORY_MAPPING.items(): if k.lower() == raw_cat.lower(): return v # 模糊匹配：提取关键词 keywords = ["phone", "mobile", "cell"] for kw in keywords: if kw in raw_cat.lower(): return "PHONE" return "OTHER" df['category_code'] = df['raw_category'].apply(normalize_category)

痛点3：用户等级的“逻辑黑洞”
users.level字段有20%空值。业务规则：若为空，则根据历史订单总金额推算——>=10000为VIP，>=5000为GOLD，其余为BRONZE。

关键洞察：这个规则不能写在Transform代码里硬编码，而应外置为配置：

# transform_rules.yaml user_level_imputation: strategy: "amount_based" thresholds: - amount: 10000.0 level: "VIP" - amount: 5000.0 level: "GOLD" default: "BRONZE"

Transform代码只读取配置，不包含业务逻辑：

def impute_user_level(row, rules): if pd.notna(row['level']): return row['level'] # 读取配置阈值 for t in rules['thresholds']: if row['total_amount'] >= t['amount']: return t['level'] return rules['default'] # 加载配置 with open("transform_rules.yaml") as f: rules = yaml.safe_load(f) df['user_level'] = df.apply(lambda r: impute_user_level(r, rules), axis=1)

踩坑心得：业务规则必须与代码分离。我曾在一个项目中把VIP阈值10000硬写进Python，结果市场部临时将门槛降到8000，运维半夜紧急发版，而如果规则外置，只需改一行YAML，重启任务即可。

3.2 转换（Transformation）：从字段到特征，构建分析友好型宽表

Transform的终极目标，是产出下游可直接消费的宽表（Wide Table）。它不是简单JOIN，而是语义融合。

我们构建sales_analytics宽表，需JOIN四张源表。但JOIN顺序有讲究：

1. 以orders为主表（分析主体是订单）
2. LEFT JOINusers（一个用户可有多单，但订单必有用户）
3. LEFT JOINorder_items（一个订单可有多商品）
4. LEFT JOINproducts（一个商品对应一个类目）

错误做法：一次性pd.merge()四表。内存爆炸，且无法处理一对多导致的行膨胀。

正确分步法：

# Step1: 订单主表 + 用户信息 df_orders_users = pd.merge( orders_df, users_df[['id', 'level']], left_on='user_id', right_on='id', how='left', suffixes=('', '_user') ) # Step2: 为每个订单项扩展商品类目 df_items_with_cat = pd.merge( order_items_df, products_df[['id', 'category_code']], left_on='product_id', right_on='id', how='left' ) # Step3: 按订单聚合商品类目（取第一个非空，或拼接） df_items_agg = df_items_with_cat.groupby('order_id')['category_code'].apply( lambda x: x.dropna().iloc[0] if len(x.dropna()) > 0 else 'OTHER' ).reset_index(name='category_code') # Step4: 主表 + 聚合后的类目 df_final = pd.merge( df_orders_users, df_items_agg, left_on='id', right_on='order_id', how='left' )

关键技巧：对category_code使用groupby().apply()而非agg('first')，因为first()会取原始顺序第一行，而dropna().iloc[0]确保取到有效值。

最终宽表结构：

3.3 质量校验（Validation）：Transform不是“做完就行”，而是“做对为止”

Transform后必须执行质量校验，这是防止垃圾进、垃圾出的最后防线。我们定义三类校验：

1. 结构校验（Schema Validation）
检查字段类型、非空约束：

def validate_schema(df: pd.DataFrame) -> bool: errors = [] if df['order_id'].isnull().sum() > 0: errors.append("order_id has null values") if not pd.api.types.is_string_dtype(df['user_level']): errors.append("user_level is not string type") if errors: print("Schema validation failed:", errors) return False return True

2. 业务校验（Business Rule Validation）
检查业务逻辑一致性：

def validate_business_rules(df: pd.DataFrame) -> bool: # VIP用户订单金额应 >= 5000 vip_orders = df[df['user_level'] == 'VIP'] low_value_vip = vip_orders[vip_orders['total_amount'] < 5000] if len(low_value_vip) > 0: print(f"Found {len(low_value_vip)} VIP orders with amount < 5000") return False return True

3. 统计校验（Statistical Validation）
对比前后数据分布：

def validate_statistics(df_before: pd.DataFrame, df_after: pd.DataFrame): # 订单总数不应减少（除非有明确过滤） if len(df_after) < len(df_before) * 0.95: # 允许5%损耗 print("Warning: Row count dropped >5%") # 用户等级分布应合理 level_dist = df_after['user_level'].value_counts(normalize=True) if level_dist.get('VIP', 0) > 0.3: # VIP占比超30%？ print("Warning: VIP ratio unusually high")

校验失败时，ETL任务必须中断，并输出清晰错误报告。我坚持一个原则：宁可任务失败，不可数据带病入库。线上报表出错一次，修复成本是预防成本的10倍。

4. Load阶段：加载不是“塞进去”，而是“交钥匙”——构建可信赖、可回滚、可审计的目标库

Load是ETL的临门一脚，也是最容易被轻视的一环。很多人以为to_sql()一下就完事，结果导致下游分析师查到一半，表结构突然变了；或者凌晨三点任务失败，第二天发现昨天的数据没了。Load的本质，是向下游交付一份可信赖的数据资产，它必须满足：一致性、可追溯、可回滚。

4.1 目标库设计：宽表不是“大杂烩”，而是有范式的分析模型

我们用SQLite模拟目标库analytics.db。关键设计原则：

• 单一事实表（Fact Table）：sales_analytics，存储订单粒度的度量值（total_amount）和维度键（user_id,category_code）
• 维度表（Dimension Tables）：独立存储维度属性，如dim_user（含user_id,level,region），避免宽表冗余
• 代理键（Surrogate Key）：为sales_analytics添加自增sk_id，作为唯一主键，隔离业务键变更影响

建表语句：

-- 目标库：analytics.db CREATE TABLE sales_analytics ( sk_id INTEGER PRIMARY KEY AUTOINCREMENT, order_id TEXT NOT NULL, created_at DATETIME NOT NULL, user_id INTEGER NOT NULL, user_level TEXT NOT NULL, category_code TEXT NOT NULL, total_amount REAL NOT NULL, etl_run_id TEXT NOT NULL, -- 关联抽取批次 load_timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ); -- 创建索引加速查询 CREATE INDEX idx_sales_date ON sales_analytics(created_at); CREATE INDEX idx_sales_user ON sales_analytics(user_id);

注意etl_run_id字段——它把每条记录绑定到一次ETL执行，是数据血缘的基石。当某天发现2024-03-15的数据异常，你只需SELECT * FROM sales_analytics WHERE etl_run_id = 'etl_20240315_0215'，即可精准定位问题批次。

4.2 加载策略：全量覆盖、增量追加，还是渐进式更新？

Load策略必须与Extract策略严格匹配：

本实践采用渐进式更新，因订单ID是天然业务主键。SQLite不支持ON CONFLICT，我们用临时表实现：

def load_incremental(df: pd.DataFrame, db_path: str, run_id: str): conn = sqlite3.connect(db_path) cursor = conn.cursor() # 1. 创建临时表，存入新数据 temp_table = "sales_analytics_temp" df.to_sql(temp_table, conn, if_exists='replace', index=False) # 2. UPSERT：新数据INSERT，旧数据UPDATE # SQLite 3.24+ 支持 INSERT ... ON CONFLICT，此处用兼容写法 cursor.execute(f""" INSERT OR REPLACE INTO sales_analytics (order_id, created_at, user_id, user_level, category_code, total_amount, etl_run_id) SELECT order_id, created_at, user_id, user_level, category_code, total_amount, '{run_id}' FROM {temp_table} """) # 3. 清理临时表 cursor.execute(f"DROP TABLE IF EXISTS {temp_table}") conn.commit() conn.close()

关键点：INSERT OR REPLACE会删除旧行再插入新行，确保etl_run_id始终为最新批次。这比单纯INSERT更安全，避免下游因重复数据产生统计偏差。

4.3 可回滚机制：没有回滚能力的ETL，就像没有刹车的汽车

任何生产ETL都必须支持一键回滚。我们的方案是：每次Load前，备份当前目标表的元数据快照。

备份内容不是全量数据（太重），而是：

• 表结构DDL（CREATE TABLE语句）
• 关键统计信息（行数、最大created_at、etl_run_id分布）
• 最近3次etl_run_id的完整数据快照（存为parquet）

备份脚本backup_before_load.py：

def backup_target_state(db_path: str, run_id: str): conn = sqlite3.connect(db_path) # 备份DDL ddl = conn.execute("SELECT sql FROM sqlite_master WHERE tbl_name='sales_analytics'").fetchone()[0] with open(f"backup/{run_id}_ddl.sql", "w") as f: f.write(ddl) # 备份统计 stats = conn.execute(""" SELECT COUNT(*) as row_count, MAX(created_at) as max_date, GROUP_CONCAT(DISTINCT etl_run_id) as run_ids FROM sales_analytics """).fetchone() with open(f"backup/{run_id}_stats.json", "w") as f: json.dump({ "row_count": stats[0], "max_date": stats[1], "run_ids": stats[2].split(',') if stats[2] else [] }, f) # 备份最近3次数据（按etl_run_id倒序取） recent_runs = conn.execute(""" SELECT DISTINCT etl_run_id FROM sales_analytics ORDER BY etl_run_id DESC LIMIT 3 """).fetchall() for (r_id,) in recent_runs: df = pd.read_sql_query(f"SELECT * FROM sales_analytics WHERE etl_run_id='{r_id}'", conn) df.to_parquet(f"backup/{run_id}_{r_id}.parquet") conn.close()

当Load失败或数据异常，执行rollback_to_run_id.py即可恢复：

python rollback_to_run_id.py --target-run-id etl_20240314_0215

该脚本会：

1. 从备份中读取etl_20240314_0215的parquet文件
2. 清空当前sales_analytics表
3. 重新加载该parquet数据
4. 更新etl_run_id为etl_20240314_0215_ROLLBACK（标记回滚）

实操心得：回滚不是救火，而是日常。我在某项目中设置每日凌晨自动备份，结果上线首周就因类目映射规则调整失误，靠回滚3小时前的快照，10分钟内恢复服务。没有回滚，就没有上线勇气。

5. 全流程串联与实战验证：用一个可运行的Python脚本，跑通从Extract到Load的每一行代码

现在，把前面所有环节串起来，形成一个完整的、可一键运行的ETL脚本。这不是玩具Demo，而是经过生产环境简化的真实骨架。你只需复制保存为etl_pipeline.py，安装依赖后即可执行。

5.1 环境准备：5分钟搭建本地ETL沙箱

# 创建虚拟环境（推荐） python -m venv etl_env source etl_env/bin/activate # Linux/Mac # etl_env\Scripts\activate # Windows # 安装核心依赖 pip install pandas numpy pyyaml sqlite3 # 创建项目目录 mkdir etl_project && cd etl_project touch etl_pipeline.py mkdir -p config data backup logs

5.2 完整可运行脚本：etl_pipeline.py

#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ETL入门全流程实践：从Extract到Load 作者：某数据工程师 版本：1.0 """ import os import sqlite3 import pandas as pd import numpy as np import yaml import json from datetime import datetime from pathlib import Path # ==================== 配置管理 ==================== CONFIG_DIR = Path("config") DATA_DIR = Path("data") BACKUP_DIR = Path("backup") LOGS_DIR = Path("logs") def load_config(config_name: str) -> dict: with open(CONFIG_DIR / config_name, "r", encoding="utf-8") as f: return yaml.safe_load(f) # ==================== Extract阶段 ==================== def extract_orders(db_path: str, last_max_time: str = None) -> pd.DataFrame: """抽取订单数据，并解析时间字段""" conn = sqlite3.connect(db_path) df = pd.read_sql_query("SELECT id, created_at, user_id, total_amount FROM orders", conn) conn.close() # 解析时间，容错处理 df['created_at_parsed'] = pd.to_datetime( df['created_at'], format='ISO8601', errors='coerce' ) if last_max_time: cutoff = pd.to_datetime(last_max_time) df = df[df['created_at_parsed'] >= cutoff].copy() return df def extract_users(db_path: str) -> pd.DataFrame: conn = sqlite3.connect(db_path) df = pd.read_sql_query("SELECT id, level FROM users", conn) conn.close() return df def extract_order_items(db_path: str) -> pd.DataFrame: conn = sqlite3.connect(db_path) df = pd.read_sql_query("SELECT order_id, product_id FROM order_items", conn) conn.close() return df def extract_products(db_path: str) -> pd.DataFrame: conn = sqlite3.connect(db_path) df = pd.read_sql_query("SELECT id, category_code FROM products", conn) conn.close() return df def run_extract() -> dict: """执行全部抽取，返回字典 {table_name: df}""" source_db = DATA_DIR / "source_db.db" # 获取上次最大时间（首次运行为空） last_time = None if (LOGS_DIR / "last_run.json").exists(): with open(LOGS_DIR / "last_run.json", "r") as f: last_info = json.load(f) last_time = last_info.get("max_created_at") orders = extract_orders(str(source_db), last_time) users = extract_users(str(source_db)) items = extract_order_items(str(source_db)) products = extract_products(str(source_db)) # 记录抽取日志 log_entry = { "run_id": f"etl_{datetime.now().strftime('%Y%m%d_%H%M')}", "start_time": datetime.utcnow().isoformat() + "Z", "tables": [ {"name": "orders", "record_count": len(orders)}, {"name": "users", "record_count": len(users)}, {"name": "order_items", "record_count": len(items)}, {"name": "products", "record_count": len(products)}, ] } LOGS_DIR.mkdir(exist_ok=True) with open(LOGS_DIR / "extract_log.json", "w") as f: json.dump(log_entry, f, indent=2) return { "orders": orders, "users": users, "items": items, "products": products } # ==================== Transform阶段 ==================== def clean_address(address: str) -> str: if not isinstance(address, str): return "" address = re.sub(r'[\u200b-\u200f\u202a-\u202e]', '', address) address = address.replace(' ', ' ') address = re.sub(r'\s+', ' ', address).strip() return address CATEGORY_MAPPING = { "MOBILE": "PHONE", "mobile": "PHONE", "Mobile Phone": "PHONE", "PHONE": "PHONE", "TABLET": "TABLET", "LAPTOP": "LAPTOP" } def normalize_category(raw_cat: str) -> str: if not raw_cat: return "OTHER" if raw_cat in CATEGORY_MAPPING: return CATEGORY_MAPPING[raw_cat] if raw_cat.lower() in [k.lower() for k in CATEGORY_MAPPING.keys()]: for k, v in CATEGORY_MAPPING.items(): if k.lower() == raw_cat.lower(): return v keywords = ["phone", "mobile", "cell"] for kw in keywords: if kw in raw_cat.lower(): return "PHONE" return "OTHER" def transform_data(extracted: dict) -> pd.DataFrame: """核心Transform逻辑""" orders = extracted["orders"] users = extracted["users"] items = extracted["items"] products = extracted["products"] # Step1: 订单 + 用户 df = pd.merge( orders, users[['id', 'level']], left_on='user_id', right_on='id', how='left', suffixes=('', '_user') ) df.rename(columns={'level': 'user_level'}, inplace=True) # Step2: 订单项 + 商品类目 items_with_cat = pd.merge( items, products[['id', 'category_code']], left_on='product_id', right_on='id', how='left' ) # Step3: 按订单聚合类目 cat_agg = items_with_cat.groupby('order_id')['category_code'].apply( lambda x: x.dropna().iloc[0] if len(x.dropna()) > 0 else 'OTHER' ).reset_index(name='category_code') # Step4: 主表 + 类目 df = pd.merge( df, cat_agg, left_on='id', right_on='order_id', how='left' ) # 填充缺失的user_level（业务规则） rules = load_config("transform_rules.yaml") def impute_level(row): if pd.notna(row['user_level']): return row['user_level'] for t in rules['user_level_imputation']['thresholds']: if row['total_amount'] >= t['amount']: return t['level'] return rules['user_level_imputation']['default'] df['user_level'] = df.apply(impute_level, axis=1) # 清洗类目 df['category_code'] = df['category_code'].apply(normalize_category) # 验证 assert validate_schema(df), "Schema validation failed" assert validate_business_rules(df), "Business rule validation failed" return df def validate_schema(df: pd.DataFrame) -> bool: if df['id'].isnull().sum() > 0: print("ERROR: order_id has nulls") return False if not pd.api.types.is_numeric_dtype(df['user_id']): print("ERROR: user_id is not numeric") return False return True def validate_business_rules(df: pd.DataFrame) -> bool: vip_low = df[(df['user_level'] == 'VIP') & (df['total_amount'] < 5000