编辑“︁ETL过程详解”︁（章节）

== ETL 详细流程 ==

=== 1. 提取（Extract） ===
从源系统抽取数据，支持全量或增量方式：
* '''全量抽取'''：首次加载时提取全部数据。
* '''增量抽取'''：仅提取自上次抽取后的变更数据（通过时间戳、日志或CDC技术实现）。

示例：从MySQL数据库提取数据（Python代码）：
<syntaxhighlight lang="python">
import pandas as pd
import pymysql

# 连接MySQL数据库
connection = pymysql.connect(
    host="localhost",
    user="user",
    password="password",
    database="source_db"
)

# 全量提取orders表数据
query = "SELECT * FROM orders"
df = pd.read_sql(query, connection)
connection.close()
print(df.head())  # 查看前5行数据
</syntaxhighlight>

'''输出示例'''：
<pre>
   order_id  customer_id  amount  order_date
0      1001           52   99.50  2023-01-15
1      1002           14  150.00  2023-01-16
</pre>

=== 2. 转换（Transform） ===
关键操作包括：
* '''数据清洗'''：处理缺失值、去重、修正格式错误。
* '''数据标准化'''：统一单位、编码（如性别"男/女"转为"M/F"）。
* '''业务规则应用'''：计算衍生字段（如销售额=单价×数量）。
* '''聚合'''：按维度汇总（如按地区统计销售总额）。

示例：使用Pandas进行数据转换：
<syntaxhighlight lang="python">
# 清洗：填充缺失值
df['amount'].fillna(0, inplace=True)

# 标准化：日期格式统一
df['order_date'] = pd.to_datetime(df['order_date'])

# 业务规则：标记大额订单
df['is_large_order'] = df['amount'] > 100

# 聚合：按客户ID统计订单数
customer_stats = df.groupby('customer_id').agg(
    total_orders=('order_id', 'count'),
    total_amount=('amount', 'sum')
)
print(customer_stats.head())
</syntaxhighlight>

'''输出示例'''：
<pre>
             total_orders  total_amount
customer_id                            
14                      1         150.0
52                      1          99.5
</pre>

=== 3. 加载（Load） ===
将处理后的数据写入目标系统，常见策略：
* '''全量加载'''：覆盖目标表所有数据。
* '''增量加载'''：仅追加新数据或更新变化记录。
* '''缓慢变化维（SCD）'''：处理维度表的历史变化（Type 1/2/3）。

示例：加载到PostgreSQL数据仓库：
<syntaxhighlight lang="python">
from sqlalchemy import create_engine

# 连接目标数据库
engine = create_engine("postgresql://user:password@localhost/warehouse")

# 全量加载到target_orders表
customer_stats.to_sql(
    "customer_stats",
    engine,
    if_exists="replace",  # 覆盖现有表
    index=True
)
</syntaxhighlight>