第03课：数据提取-API-数据库与文件

📥 数据提取-API-数据库与文件

数据提取是ETL第一步，负责从各种异构数据源中获取原始数据。优秀的提取策略需要平衡完整性、时效性和资源消耗。

🐍 数据库数据提取

# 数据库提取（SQLite实机）
import sqlite3, time, random
from datetime import datetime, timedelta

conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
c = conn.cursor()
c.execute('CREATE TABLE orders (id INTEGER PRIMARY KEY AUTOINCREMENT, customer_id TEXT, amount REAL, status TEXT, region TEXT, created_at TIMESTAMP)')
regions = ['华东','华南','华北','西南','东北']
for i in range(1000):
    c.execute("INSERT INTO orders VALUES (?,?,?,?,?,?)",
        (None, f"客户_{random.randint(1,200)}", round(random.uniform(10,5000),2),
         random.choice(['pending','shipped','delivered']), random.choice(regions),
         (datetime.now()-timedelta(days=random.randint(0,30))).isoformat()))
conn.commit()
print(f"测试库: 1000订单")

# 全量提取
start = time.perf_counter()
rows = [dict(r) for r in c.execute("SELECT * FROM orders").fetchall()]
print(f"全量: {len(rows)}条, {(time.perf_counter()-start)*1000:.2f}ms")

# 增量提取(id>990)
start = time.perf_counter()
rows = [dict(r) for r in c.execute("SELECT * FROM orders WHERE id > 990").fetchall()]
print(f"增量(id>990): {len(rows)}条, {(time.perf_counter()-start)*1000:.2f}ms")

# 并行提取(4分区)
for s,e in [(1,250),(251,500),(501,750),(751,1000)]:
    rows = c.execute("SELECT COUNT(*) FROM orders WHERE id BETWEEN ? AND ?", (s,e)).fetchone()[0]
    print(f"  分区[{s},{e}]: {rows}条")

print("✅ 验证通过 - 数据库提取运行正常")

📐 提取性能优化

# 文件提取：CSV/JSON
import csv, json, io, random

class FileExtractor:
    def extract_csv(self, content, delimiter=','):
        reader = csv.DictReader(io.StringIO(content), delimiter=delimiter)
        return [dict((k, float(v) if v.replace('.','',1).isdigit() else v.strip())
            for k,v in row.items() if v.strip()) for row in reader]
    def extract_json(self, content):
        data = json.loads(content)
        if isinstance(data, list): return data
        for key in ['data','records','items']:
            if key in data and isinstance(data[key], list): return data[key]
        return [data]

csv_content = "id,name,amount,category\n"
for i in range(1, 51):
    csv_content += f"{i},商品_{i},{round(random.uniform(10,500),2)},{random.choice(['A','B','C'])}\n"

json_content = json.dumps({'data':[
    {'id':i,'sensor':f"sensor_{random.randint(1,20)}",'value':round(random.uniform(0,100),2)}
    for i in range(1,31)]}, ensure_ascii=False)

ext = FileExtractor()
csv_r = ext.extract_csv(csv_content)
json_r = ext.extract_json(json_content)
print(f"CSV: {len(csv_r)}条, JSON: {len(json_r)}条")
print(f"CSV示例: {csv_r[0]}")
print("✅ 验证通过 - 文件提取运行正常")

📖 原理深入：ETL设计核心原则

🏆 最佳实践：生产环境ETL检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答