第2章数据源分析与连接

2、数据源分析与连接：关系型数据库、文件数据源、API数据源、日志数据采集

大家好，我是老赵。做ETL这么多年，我最大的感触就是：数据源连接是地基。地基没打好，后面跑得再快也是白搭。今天咱们就聊聊这四种最常见的数据源——关系型数据库、文件、API和日志。我会把我在项目中踩过的坑、总结的经验，一股脑儿倒出来。

核心观点：连接数据源不是简单的“配个JDBC串”或“读个文件”。你要考虑的是：连接稳定性、增量抽取策略、数据一致性校验、以及异常恢复机制。说白了，就是“连得上、抽得稳、对得上”。

2.1 关系型数据库：MySQL与PostgreSQL

关系型数据库是ETL的“老本行”。我个人习惯把MySQL和PostgreSQL分开对待，虽然它们都支持JDBC，但性格差异挺大。

2.1.1 MySQL连接要点

MySQL在ETL场景下，最让人头疼的是主从延迟。我记得有一次做实时同步，从库延迟了30秒，结果ETL抽到的数据跟主库对不上，业务方直接炸了。所以我的建议是：

优先读从库，但必须监控延迟时间。延迟超过阈值就切主库。
使用批量读取，别一条一条查。MySQL的游标机制在大量数据下会撑爆内存。
增量字段选择：最好用自增ID或更新时间戳。我曾经见过有人用“最后修改时间”做增量，结果同一秒内修改了两条数据，漏了一条。嗯，这里要注意。

-- 一个稳妥的增量抽取SQL示例
SELECT * FROM orders 
WHERE update_time > '2024-01-01 00:00:00' 
AND update_time <= '2024-01-02 00:00:00'
ORDER BY id ASC
LIMIT 1000;

我的小技巧：在MySQL连接串里加上 rewriteBatchedStatements=true，批量写入性能能提升3-5倍。这个参数很多人不知道。

2.1.2 PostgreSQL连接要点

PostgreSQL在ETL里其实比MySQL更“友好”。为什么？因为它有COPY命令。你想想看，用COPY导入几十万行数据，比逐条INSERT快了一个数量级。

但PostgreSQL有个坑——事务ID回卷。我曾经在维护一个跑了三年的ETL任务时，突然发现PostgreSQL拒绝写入，报错说“事务ID已耗尽”。当时我冷汗都下来了。后来才知道，需要定期执行 VACUUM。

# 使用psycopg2的copy_from方法
import psycopg2

conn = psycopg2.connect("dbname=test user=postgres")
cur = conn.cursor()
with open('data.csv', 'r') as f:
    cur.copy_from(f, 'orders', sep=',')
conn.commit()

避坑指南：PostgreSQL的默认隔离级别是“读已提交”，但在高并发写入时，可能会出现“不可重复读”。如果你的ETL需要精确的一致性快照，记得设置 SET TRANSACTION ISOLATION LEVEL SERIALIZABLE。

2.2 文件数据源：CSV、JSON、XML

文件数据源看起来简单，其实最容易出幺蛾子。我见过太多因为文件编码、分隔符、换行符导致ETL跑飞的情况。

2.2.1 CSV文件

CSV是“最熟悉的陌生人”。你以为它简单？其实它没有统一标准。我总结了几条铁律：

永远不要相信CSV的列数。数据里可能包含逗号、换行符、引号。用Python的 csv 模块或Pandas的 read_csv，别自己写解析。
指定编码。UTF-8 with BOM 和 UTF-8 是两码事。我遇到过因为BOM头导致第一列字段名多了一个不可见字符，排查了整整一天。
大文件处理：别一次性读入内存。用迭代器或分块读取。

import pandas as pd

# 分块读取大CSV
chunk_size = 10000
for chunk in pd.read_csv('large_file.csv', chunksize=chunk_size, encoding='utf-8-sig'):
    # 处理每个chunk
    process_chunk(chunk)

2.2.2 JSON与XML

JSON和XML都是嵌套结构，ETL的核心是展平。说白了，就是把嵌套的层级变成扁平的表格。

我个人习惯用 json_normalize 处理JSON，用 xml.etree.ElementTree 处理XML。但要注意：

JSON的数组嵌套：如果数组里套数组，展平后会产生笛卡尔积，数据量会爆炸。我曾经处理过一个订单JSON，一个订单里有100个商品，每个商品又有10个属性，展平后变成了1000行。嗯，这里要提前评估。
XML的命名空间：很多新手会忽略命名空间，导致解析不到数据。记得用 findall('ns:tag', namespaces)。

import json
from pandas import json_normalize

data = json.load(open('orders.json'))
# 展平嵌套JSON
df = json_normalize(data, record_path='items', meta=['order_id', 'customer_name'])

2.3 API数据源

API数据源是现在ETL的“新常态”。但API连接有个核心问题——限流和重试。

我记得有一次对接一个第三方支付API，对方限流策略是每分钟100次。我的ETL任务没做限流控制，结果跑了5分钟就被封IP了。后来我学乖了，所有API调用都加上重试机制和退避策略。

import requests
from time import sleep

def call_api_with_retry(url, max_retries=3):
    for i in range(max_retries):
        try:
            resp = requests.get(url, timeout=10)
            if resp.status_code == 429:  # 限流
                sleep(2 ** i)  # 指数退避
                continue
            resp.raise_for_status()
            return resp.json()
        except Exception as e:
            print(f"第{i+1}次重试失败: {e}")
            sleep(1)
    raise Exception("API调用失败")

我的经验：API数据源最好做两层缓存。第一层是内存缓存（比如LRU），第二层是磁盘缓存（比如SQLite）。这样即使API挂了，ETL也能从缓存里拿数据，不至于断流。

2.4 日志数据采集

日志数据采集，说白了就是“从文本里捞金”。但日志的格式千奇百怪，没有统一的Schema。

我常用的工具有两个：Filebeat 和 Fluentd。Filebeat轻量，适合单机采集；Fluentd功能强，适合复杂路由。

但不管用哪个，核心都是解析日志格式。我建议用正则表达式或Grok模式。举个例子：

# 一个典型的Nginx日志解析
%{IPORHOST:client_ip} - %{USERNAME:remote_user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}" %{NUMBER:status} %{NUMBER:body_bytes} "%{URI:referrer}" "%{DATA:user_agent}"

避坑指南：日志采集最怕的是日志轮转。我曾经因为没处理好日志轮转，导致Filebeat重复读取了同一个日志文件，数据翻了一倍。解决方案是：使用 harvester 的 close_inactive 参数，并开启 clean_removed。

知识体系总览

下面这张图是我自己画的，把本章的知识点串了起来。你仔细看看，就能明白数据源连接的全貌。

好了，这一章的内容就到这里。数据源连接是ETL的“第一公里”，也是最容易出问题的地方。你把这些基础打牢了，后面做数据清洗、转换、加载，才能游刃有余。

📚 课程目录

2、数据源分析与连接：关系型数据库、文件数据源、API数据源、日志数据采集

2.1 关系型数据库：MySQL与PostgreSQL

2.1.1 MySQL连接要点

2.1.2 PostgreSQL连接要点

2.2 文件数据源：CSV、JSON、XML

2.2.1 CSV文件

2.2.2 JSON与XML

2.3 API数据源

2.4 日志数据采集

知识体系总览