数据源接入：把管道连到真实世界

管道搭好了，但数据从哪来？
这才是元数据管道的真正起点。我个人觉得，数据源接入是整套体系里最烦人、也最关键的一环——你想想看，后面所有清洗、转换、建模，都建立在接入这一步的可靠性上。接错了，管道就是空中楼阁。

这一章我们专门聊四种最常见的数据源：关系型数据库、NoSQL、文件系统和API。每种我都会给出一套经过实战检验的接入模式，附带具体的Python代码片段和避坑清单。

核心原则：统一接口 + 可配置连接 + 异常重试 + 增量标记。所有数据源都抽象成同一个 DataSource 基类，业务层只管调用 fetch() 和 schema()。

关系型数据库：SQL 是管道的老朋友

关系型数据库（MySQL / PostgreSQL / SQL Server）是元数据最常见的源头。大多数企业的核心业务数据都躺在关系库里。我自己的经验是，接入关系库最怕两件事：慢查询拖死管道，以及连接泄露。

推荐方案：SQLAlchemy + Pandas

Python 生态里，SQLAlchemy 是事实标准。它天然支持连接池，配合 Pandas 的 read_sql 能把结果直接转成 DataFrame，省去逐行解析的麻烦。

from sqlalchemy import create_engine, text
import pandas as pd

def fetch_from_mysql(query: str, connection_string: str) -> pd.DataFrame:
    engine = create_engine(connection_string, pool_size=5, max_overflow=2)
    with engine.connect() as conn:
        df = pd.read_sql(text(query), conn)
    engine.dispose()
    return df

# 示例用法
dsn = "mysql+pymysql://user:pass@host:3306/mydb"
df = fetch_from_mysql("SELECT * FROM orders WHERE updated_at > '2024-01-01'", dsn)
小技巧：连接字符串里不要硬编码密码。我习惯用环境变量 + os.getenv，或者用 AWS Secrets Manager / HashiCorp Vault。
参数说明推荐值
pool_size连接池大小5~10（视并发而定）
max_overflow超出池的最大临时连接数2~5
pool_pre_ping每次取连接时检查有效性True

增量读取的策略

全量读取只适合 T-1 离线批处理。实时管道必须走增量。我常用的模式是维护一张 pipeline_bookmark 表，记录每个表最后提取的 updated_at 或自增 ID。

-- pipeline_bookmark 表结构
CREATE TABLE pipeline_bookmark (
    source_table VARCHAR(255) PRIMARY KEY,
    last_cursor VARCHAR(64),
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 每次提取时先读书签，再构造条件
SELECT * FROM orders
WHERE id > :last_id
ORDER BY id ASC
LIMIT 10000;
踩坑回忆：我曾经直接用了 SELECT * 全量覆盖，结果某次源表误操作插入了一亿条脏数据，管道直接崩溃。后来强制限定分页 + 书签，再也没出过类似问题。

NoSQL：MongoDB / Redis 的接入姿势

NoSQL 跟关系库不一样，没有统一的 SQL 接口。MongoDB 文档嵌套深，Redis 更强调键值查找。不过抽象层思路一样：写一个适配器，对外暴露同样的 fetch 和 schema 方法。

MongoDB 示例

from pymongo import MongoClient
import pandas as pd

def fetch_from_mongo(
    connection_uri: str,
    database: str,
    collection: str,
    filter_query: dict
) -> pd.DataFrame:
    client = MongoClient(connection_uri, maxPoolSize=10)
    db = client[database]
    coll = db[collection]
    cursor = coll.find(filter_query).batch_size(1000)
    docs = list(cursor)
    client.close()
    return pd.DataFrame(docs)

# 使用
df = fetch_from_mongo(
    "mongodb://user:pass@host:27017",
    "myapp",
    "events",
    {"event_type": "purchase", "timestamp": {"$gte": start_ts}}
)

注意 MongoDB 的文档可能包含 ObjectId 等不可 JSON 序列化的字段。我一般在返回前做一次类型转换：df['_id'] = df['_id'].astype(str)。

Redis 接入（缓存元数据）

Redis 通常不存全量数据，而是放元数据缓存（比如表结构快照、最新偏移量）。但有时候也得直接从 Redis 里拉取键列表。下面是一个简单的批量扫描模式：

import redis

r = redis.Redis(host='redis-host', port=6379, decode_responses=True)
cursor = 0
keys = []
while True:
    cursor, batch = r.scan(cursor=cursor, match='meta:*', count=500)
    keys.extend(batch)
    if cursor == 0:
        break

# 批量获取值
pipeline = r.pipeline()
for key in keys:
    pipeline.get(key)
values = pipeline.execute()
注意：不要在 Redis 上用 KEYS *，线上会阻塞。使用 SCAN 是唯一正确的方式。

文件系统：CSV / Parquet / 日志文件

文件系统是最原始的数据源，但也是最常见的一种——尤其是 HDFS、S3、MinIO 这类对象存储。管道的任务是把散落的文件统一注册为元数据记录。

统一文件发现与模式推导

我习惯先扫描指定路径，识别文件类型（CSV / JSON / Parquet / Avro），然后自动推断模式。下面是一个简化版：

import os
import pyarrow as pa
import pyarrow.parquet as pq

def discover_files(root_path: str, suffix: str = ".parquet"):
    for dirpath, _, filenames in os.walk(root_path):
        for fn in filenames:
            if fn.endswith(suffix):
                yield os.path.join(dirpath, fn)

def infer_schema(file_path: str):
    # 只读头几条记录推断列类型
    table = pq.read_table(file_path, nrows=100)
    return table.schema

for f in discover_files("/data/events/"):
    schema = infer_schema(f)
    print(f, schema)
文件格式读取库模式推断压缩支持
CSVPandas / csv.DictReader需要手动或采样推断gzip / bz2
JSON / JSONL或者 json 模块可自动嵌套展开gzip
ParquetPyArrow / FastParquet列式 Schema 自带Snappy / Gzip / Zstd
Avrofastavro / avroSchema 嵌入文件Deflate
避坑指南：我曾经直接读取一个 100GB 的 CSV 全部加载到内存，结果 OOM。后来改用 pandas.read_csv(chunksize=10000) 或者用 PyArrow 做流式读取。文件系统接入一定要关注文件大小和行数，分片处理。

API 数据源：HTTP 协议的通用接入

API 接入最灵活，但也最不可靠。网络超时、限流、认证过期、响应格式变化——每个都是潜在的坑。我总结了三条铁律：重试 + 退避 + 超时控制。

通用 REST API 接入模板

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

class ApiDataSource:
    def __init__(self, base_url, token):
        self.base_url = base_url.rstrip('/')
        self.headers = {"Authorization": f"Bearer {token}"}
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=30))
    def fetch(self, endpoint: str, params: dict = None):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        resp = self.session.get(url, params=params, timeout=30)
        resp.raise_for_status()
        return resp.json()

# 用法
api = ApiDataSource("https://api.example.com/v2", os.getenv("API_TOKEN"))
data = api.fetch("users", {"page": 1, "per_page": 50})

很多 API 采用分页。我习惯封装一个分页迭代器，自动遍历所有页：

def fetch_all_pages(api, endpoint, page_param="page", start=1):
    page = start
    while True:
        batch = api.fetch(endpoint, {page_param: page})
        if not batch:
            break
        yield from batch
        page += 1

限流处理

第三方 API 经常有速率限制（比如每分钟 100 次）。我常用 ratelimit 库或者简单的 time.sleep 来减速。

from time import sleep
from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=80, period=60)
def safe_fetch(api, endpoint):
    return api.fetch(endpoint)

# 调用时自动限流
data = safe_fetch(api, "reports")
总结：无论源端是哪种系统，接入层的核心都一样——
统一接口 + 可配置连接信息 + 异常重试 + 增量标记。
下面这张图总结了四种数据源的接入架构与通用抽象层。
数据源接入架构 关系型数据库 MySQL / PostgreSQL NoSQL MongoDB / Redis 文件系统 CSV / Parquet / S3 API 数据源 REST / GraphQL 统一接入层 DataSource 抽象基类 fetch() · schema() 异常重试 · 连接池 配置驱动 · 书签管理 元数据管道 清洗 → 转换 → 建模 说明： 每种数据源由适配器实现 DataSource 接口，接入层对外暴露一致的 fetch() 与 schema() 方法。 管道上流只依赖接口，不关心底层是 SQL 还是文件扫描。

嗯，以上就是第四章节的全部内容。四种数据源各自有脾气，但只要把接入层抽象好、配置分离、重试策略到位，管道就能稳定运行。下一章我们会聊管道的数据清洗与模式对齐，到时候会用到今天接进来的这些原始数据。

嗯，以上就是第四章节的全部内容。四种数据源各自有脾气，但只要把接入层抽象好、配置分离、重试策略到位，管道就能稳定运行。

← 上一章 📖 返回目录 下一章 →