数据工程变现教程：ETL+数据管道+数据仓库

为什么学数据工程

商业价值：

• 数据工程师：¥25K-60K/月
• 数据管道搭建：¥10000-50000/项目
• 数据仓库建设：¥50000-200000/项目
• 数据治理咨询：¥5000/天

市场需求：

• 每个有数据的公司都需要数据工程
• AI时代数据量暴增，需求更大
• 数据工程师缺口巨大

核心技术栈

1. ETL工具

# 用Python做ETL
import pandas as pd

def extract(source: str) -> pd.DataFrame:
    """从数据源提取数据"""
    if source.endswith(".csv"):
        return pd.read_csv(source)
    elif source.endswith(".json"):
        return pd.read_json(source)
    else:
        # 数据库
        import sqlite3
        conn = sqlite3.connect(source)
        return pd.read_sql("SELECT * FROM data", conn)

def transform(df: pd.DataFrame) -> pd.DataFrame:
    """数据清洗和转换"""
    # 去重
    df = df.drop_duplicates()
    
    # 处理缺失值
    df = df.fillna({"name": "Unknown", "price": 0})
    
    # 类型转换
    df["date"] = pd.to_datetime(df["date"])
    df["price"] = df["price"].astype(float)
    
    # 添加计算列
    df["year"] = df["date"].dt.year
    df["month"] = df["date"].dt.month
    
    return df

def load(df: pd.DataFrame, target: str):
    """加载到目标"""
    import sqlite3
    conn = sqlite3.connect(target)
    df.to_sql("clean_data", conn, if_exists="replace", index=False)

# ETL管道
def etl_pipeline(source: str, target: str):
    df = extract(source)
    df = transform(df)
    load(df, target)
    print(f"ETL完成: {len(df)}条数据")

2. 数据管道（Airflow）

# 安装: pip install apache-airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "data-team",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "daily_etl",
    default_args=default_args,
    schedule_interval="0 2 * * *",  # 每天凌晨2点
    start_date=datetime(2026, 1, 1),
)

def extract_task():
    df = extract("source.db")
    df.to_parquet("/tmp/raw_data.parquet")

def transform_task():
    df = pd.read_parquet("/tmp/raw_data.parquet")
    df = transform(df)
    df.to_parquet("/tmp/clean_data.parquet")

def load_task():
    df = pd.read_parquet("/tmp/clean_data.parquet")
    load(df, "warehouse.db")

extract_op = PythonOperator(task_id="extract", python_callable=extract_task, dag=dag)
transform_op = PythonOperator(task_id="transform", python_callable=transform_task, dag=dag)
load_op = PythonOperator(task_id="load", python_callable=load_task, dag=dag)

extract_op >> transform_op >> load_op

3. 数据质量检查

def data_quality_check(df: pd.DataFrame) -> dict:
    """数据质量检查"""
    issues = []
    
    # 检查缺失值
    missing = df.isnull().sum()
    for col, count in missing.items():
        if count > 0:
            pct = count / len(df) * 100
            if pct > 5:
                issues.append(f"列'{col}'缺失{count}条({pct:.1f}%)")
    
    # 检查重复
    dupes = df.duplicated().sum()
    if dupes > 0:
        issues.append(f"重复数据{dupes}条")
    
    # 检查异常值
    for col in df.select_dtypes(include=[float, int]).columns:
        q1 = df[col].quantile(0.25)
        q3 = df[col].quantile(0.75)
        iqr = q3 - q1
        outliers = ((df[col] < q1 - 1.5 * iqr) | (df[col] > q3 + 1.5 * iqr)).sum()
        if outliers > 0:
            issues.append(f"列'{col}'有{outliers}个异常值")
    
    return {
        "total_rows": len(df),
        "total_cols": len(df.columns),
        "issues": issues,
        "quality_score": max(0, 100 - len(issues) * 10),
    }

商业化

数据管道服务

服务内容：
├── 数据源接入：¥3000-5000/个
├── ETL管道搭建：¥5000-15000/个
├── 数据仓库建设：¥20000-100000
├── 数据质量监控：¥3000/月
└── 数据治理咨询：¥5000/天

数据工具SaaS

产品：DataPipe — 无代码数据管道
功能：
├── 可视化数据流设计
├── 100+数据源连接器
├── 自动调度
├── 数据质量监控
└── 团队协作

定价：
├── 免费版：3个管道
├── 专业版：$49/月
├── 企业版：$199/月

月收入预估

数据工程服务：
├── 2个项目/月 × ¥15000 = ¥30000/月
├── 维护客户 × ¥5000/月 = ¥10000
└── 总计：¥40000/月

数据工具SaaS（50用户）：
├── 月收入：¥10000-30000
└── 成本：¥1000/月

你已有SQLite+Python数据处理能力——升级到企业级数据工程不难。