Prompt 驱动的 ETL 流程自动化实战指南

技巧点分析：

• 安全设计：密码设为变量，避免泄露；分批读取适合百万级数据。
• 用户体验：添加进度提示和数据预览，方便验证。
• 异常处理：覆盖连接失败、查询报错场景。

场景 2：从 API 接口抽取数据

对于 API 数据，稳定性至关重要。我们需要在 Prompt 中强调重试机制和超时设置。

代码实现思路： 使用 requests 库，配合循环遍历分页，并在请求失败时增加重试逻辑。

import requests
import time
import csv
from typing import List, Dict

def extract_api_products():
    API_URL = "https://api.ecommerce.com/v1/products"
    APP_KEY = "your_app_key"
    CATEGORY_ID = 3
    PAGE_SIZE = 20
    TIMEOUT = 10
    RETRY_TIMES = 2
    RETRY_INTERVAL = 3
    OUTPUT_CSV = "./products_electronic.csv"
    all_products: List[Dict] = []
    current_page = 1
    total_page = 1

    print(f"开始抽取分类 {CATEGORY_ID}（电子产品）商品数据")
    try:
        while current_page <= total_page:
            params = {
                "app_key": APP_KEY,
                "category_id": CATEGORY_ID,
                "page": current_page,
                "page_size": PAGE_SIZE
            }
            retry_count = 0
            while retry_count < RETRY_TIMES:
                try:
                    response = requests.get(API_URL, params=params, timeout=TIMEOUT)
                    response.raise_for_status()
                    result = response.json()
                    if result.get("code") != 200:
                        raise Exception(f"API 响应错误：code={result.get('code')}")
                    page_products = result.get("data", [])
                    total_page = result.get("total_page", 1)
                    all_products.extend(page_products)
                    print(f"第 {current_page} 页抽取成功，获取 {len(page_products)} 条数据")
                    break
                except Exception as e:
                    retry_count += 1
                    if retry_count >= RETRY_TIMES:
                        raise Exception(f"第 {current_page} 页抽取失败：{str(e)}")
                    print(f"第 {current_page} 页抽取失败，{RETRY_INTERVAL} 秒后重试")
                    time.sleep(RETRY_INTERVAL)
            current_page += 1

        if all_products:
            fieldnames = all_products[0].keys()
            with open(OUTPUT_CSV, "w", encoding="utf-8", newline="") as f:
                writer = csv.DictWriter(f, fieldnames=fieldnames)
                writer.writeheader()
                writer.writerows(all_products)
            print(f"\n抽取完成，共获取 {len(all_products)} 条商品数据")
        return all_products
    except Exception as e:
        print(f"\n抽取过程异常终止：{str(e)}")
        return None

if __name__ == "__main__":
    extract_api_products()

四、转换阶段（Transform）实战

转换是 ETL 中最复杂的环节，涉及清洗、计算、关联等。

场景 1：数据清洗（缺失值、异常值）

Prompt 指令要点：

• 明确缺失值填充规则（如'未知用户'）。
• 定义异常值修正逻辑（如金额<0 修正为 0）。
• 要求保留原始数据副本，新增标记字段。

代码实现：

import pandas as pd
import numpy as np

def transform_clean_orders(df_orders: pd.DataFrame) -> pd.DataFrame:
    df_orders_clean = df_orders.copy()
    print("=== 开始订单数据清洗 ===")

    # 1. 统计缺失值
    missing_before = df_orders_clean.isnull().sum()
    print(f"清洗前缺失值统计：{missing_before[missing_before > 0]}")

    # 2. 缺失值处理
    # user_id 缺失：用'未知用户_+order_id'填充
    user_id_missing_mask = df_orders_clean["user_id"].isnull()
    df_orders_clean.loc[user_id_missing_mask, "user_id"] = (
        "未知用户_" + df_orders_clean.loc[user_id_missing_mask, "order_id"]
    )
    # order_time 缺失：填充为未来时间
    order_time_missing_mask = df_orders_clean["order_time"].isnull()
    df_orders_clean.loc[order_time_missing_mask, "order_time"] = "2099-12-31 23:59:59"

    # 3. 异常值处理
    df_orders_clean["amount_flag"] = "正常"
    neg_mask = df_orders_clean["amount"] < 0
    if neg_mask.sum() > 0:
        df_orders_clean.loc[neg_mask, "amount"] = 0
        df_orders_clean.loc[neg_mask, "amount_flag"] = "异常 - 金额为负"
    high_mask = df_orders_clean["amount"] > 10000
    if high_mask.sum() > 0:
        df_orders_clean.loc[high_mask, "amount"] = 10000
        df_orders_clean.loc[high_mask, "amount_flag"] = "异常 - 超最高限额"

    print("=== 订单数据清洗完成 ===")
    return df_orders_clean

场景 2：多表关联

Prompt 指令要点：

• 明确关联键（如 user_id）和关联方式（内连接）。
• 定义派生字段计算逻辑（如用户资历天数）。

代码实现：

import pandas as pd

def transform_join_tables(df_orders_transformed: pd.DataFrame, df_users: pd.DataFrame) -> pd.DataFrame:
    print("=== 开始订单表与用户表关联 ===")
    
    # 1. 预处理：用户表时间转换
    df_users_processed = df_users.copy()
    df_users_processed["register_time"] = pd.to_datetime(
        df_users_processed["register_time"], format="%Y-%m-%d"
    )

    # 2. 内连接
    df_joined = pd.merge(
        left=df_orders_transformed,
        right=df_users_processed,
        on="user_id",
        how="inner",
        suffixes=("", "_user")
    )

    # 3. 新增派生字段
    df_joined["user_seniority"] = (
        df_joined["order_time"] - df_joined["register_time"]
    ).dt.days
    df_joined.loc[df_joined["user_seniority"] < 0, "user_seniority"] = 0

    def get_level_discount(level):
        if level == "钻石": return 0.05
        elif level == "黄金": return 0.03
        else: return 0.00

    df_joined["user_level_discount"] = df_joined["user_level"].apply(get_level_discount)
    df_joined["user_level_discount_str"] = df_joined["user_level_discount"].apply(lambda x: f"{x*100:.1f}%")

    print("=== 关联完成 ===")
    return df_joined

五、加载阶段（Load）实战

加载阶段需将数据写入目标存储，常见包括文件（CSV/Parquet）和数据库（MySQL/Hive）。

场景 1：加载到 CSV/Parquet 文件

Prompt 指令要点：

• 指定文件名规则（含日期）。
• 要求自动创建目录，失败时清理不完整文件。
• 校验写入后的数据完整性。

代码实现：

import pandas as pd
import os
from datetime import datetime
import pyarrow as pa
import pyarrow.parquet as pq

def load_to_files(df_joined: pd.DataFrame) -> bool:
    print("=== 开始将数据加载到文件 ===")
    load_success = False
    try:
        current_date = datetime.now().strftime("%Y%m%d")
        output_dir = "./etl_output/"
        csv_filename = f"orders_joined_{current_date}.csv"
        parquet_filename = f"orders_joined_{current_date}.parquet"
        csv_path = os.path.join(output_dir, csv_filename)
        parquet_path = os.path.join(output_dir, parquet_filename)

        # 创建目录
        if not os.path.exists(output_dir):
            os.makedirs(output_dir, exist_ok=True)

        # 写入 CSV
        df_joined.to_csv(path_or_buf=csv_path, index=False, encoding="utf-8", na_rep="NA")
        print(f"CSV 文件写入完成")

        # 写入 Parquet
        table = pa.Table.from_pandas(df_joined)
        pq.write_to_dataset(table, root_path=parquet_path, compression="snappy", overwrite=True)
        print(f"Parquet 文件写入完成")

        load_success = True
    except Exception as e:
        print(f"加载失败：{str(e)}")
        # 清理失败文件
        if os.path.exists(csv_path): os.remove(csv_path)
        if os.path.exists(parquet_path): shutil.rmtree(parquet_path)
    return load_success

场景 2：加载到 MySQL 数据库

Prompt 指令要点：

• 定义表结构（字段类型、主键）。
• 要求批量插入优化性能。
• 加载前后校验行数一致性。

代码实现：

import pandas as pd
from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError

def load_to_mysql(df_joined: pd.DataFrame) -> bool:
    print("=== 开始将数据加载到 MySQL 数据库 ===")
    load_success = False
    engine = None
    try:
        db_config = {
            "host": "192.168.1.100",
            "port": 3306,
            "db": "ecommerce_etl",
            "user": "etl_user",
            "password": "etl_password",
            "charset": "utf8mb4"
        }
        table_name = "orders_joined"
        conn_str = f"mysql+pymysql://{db_config['user']}:{db_config['password']}@{db_config['host']}:{db_config['port']}/{db_config['db']}?charset={db_config['charset']}"
        engine = create_engine(conn_str)

        # 创建表
        create_table_sql = f"""
            CREATE TABLE IF NOT EXISTS {table_name} (
                order_id VARCHAR(20) NOT NULL PRIMARY KEY,
                user_id VARCHAR(20) NOT NULL,
                amount DECIMAL(10,2) NOT NULL,
                order_time DATETIME NOT NULL
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
        """
        with engine.connect() as conn:
            conn.execute(create_table_sql)
            conn.execute(f"TRUNCATE TABLE {table_name};")
            df_joined.to_sql(name=table_name, con=engine, if_exists="append", index=False, chunksize=1000, method="multi")

        # 校验
        with engine.connect() as conn:
            result = conn.execute(f"SELECT COUNT(*) AS total FROM {table_name}.").fetchone()
            if result["total"] == len(df_joined):
                load_success = True
                print("加载校验成功")
    except SQLAlchemyError as e:
        print(f"数据库错误：{str(e)}")
    finally:
        if engine is not None:
            engine.dispose()
    return load_success

六、完整 ETL 流程模板

将上述逻辑整合，可形成一套完整的 Prompt 模板，便于后续调用。

Prompt 模板示例：

请作为数据工程师，生成一套完整的 ETL 自动化 Python 代码，处理'电商订单数据'，具体要求如下：
### 一、数据源信息
1. 订单数据源：MySQL 数据库...
2. 用户数据源：CSV 文件...
### 二、ETL 各阶段要求
#### （一）抽取阶段...（分批读取、异常处理）
#### （二）转换阶段...（清洗、计算、关联）
#### （三）加载阶段...（MySQL 写入、校验）
### 三、代码输出要求
1. 生成完整 Python 代码，包含函数封装...
2. 敏感信息设为变量...

基于此模板生成的代码将包含 extract_order_data, transform_data, load_data 等模块，并具备完整的日志打印和异常捕获机制。

七、优势与注意事项

核心优势

1. 降低技术门槛：非技术人员可通过自然语言定义规则。
2. 提升迭代效率：业务规则变化时，仅需修改 Prompt。
3. 场景适配灵活：擅长处理半结构化数据。

关键注意事项

1. 敏感信息保护：Prompt 中避免直接填写密码，使用占位符。
2. 数据质量校验：LLM 生成的代码可能存在逻辑漏洞，务必小批量测试。
3. 性能优化边界：超大规模数据需手动优化批量大小或索引。
4. 模型选型适配：复杂流程建议使用 ChatGPT-4 等强模型，开源模型需搭配 Few-Shot 示例。

八、课后练习

基础练习

修改 Prompt 指令，实现'从 CSV 订单数据中抽取 2024 年 2 月的未支付订单，清洗金额异常值（>5000 标记为'待审核'），并加载到 Excel 文件'。

进阶实战

设计 Prompt 并生成代码，实现'多数据源（MySQL 订单表 + API 用户表 + CSV 商品表）的 ETL 自动化，最终加载到 Hive 数据仓库'。

通过以上练习，可深入掌握 Prompt 与 ETL 流程的结合逻辑，逐步实现从'手动编写 ETL 代码'到'自然语言驱动自动化'的转型。