3. 数据采集与接入：批式采集、流式采集与数据源适配器设计
数据采集是整个数据底座的「水龙头」。水龙头开得好不好，直接决定了后面模型迭代的水质和水量。我见过太多项目，模型训练到一半才发现数据源不稳定、格式不统一，最后不得不返工重做数据管道。这一章，我们就来聊聊怎么把这个水龙头拧好。
3.1 批式采集：Sqoop 与 DataX
批式采集，说白了就是「定时搬砖」。每天凌晨把业务库的数据全量或增量地搬到数仓里。这种模式适合对实时性要求不高的场景，比如日报、周报、离线训练样本生成。
3.1.1 Sqoop：传统 Hadoop 生态的搬运工
Sqoop 是 Apache 旗下的老牌工具，专门用来在关系型数据库和 Hadoop 之间传数据。我个人习惯用它做全量导入，因为配置简单，一条命令就能搞定。
# 从 MySQL 导入用户表到 Hive
sqoop import \
--connect jdbc:mysql://192.168.1.100:3306/biz_db \
--username root \
--password secret \
--table user_profile \
--hive-import \
--hive-table dwd.user_profile_daily \
--target-dir /user/hive/warehouse/dwd.db/user_profile_daily \
--fields-terminated-by '\001' \
--num-mappers 4这里有个坑：--num-mappers 不是越大越好。我曾经在一个 8 核的 MySQL 实例上设了 16 个 mapper，结果数据库连接池直接被打满，业务方投诉了半小时。建议根据数据库的 max_connections 和 IO 能力来调，一般 4~8 个就够。
--incremental lastmodified 参数，结果每天全量跑一次，把源库的 binlog 撑爆了。增量采集一定要用 --check-column 指定时间戳字段，并配合 --last-value 使用。
3.1.2 DataX：阿里系的批式利器
DataX 是阿里巴巴开源的异构数据源同步工具。相比 Sqoop，它的插件体系更丰富，支持 MySQL、Oracle、HDFS、Hive、HBase 等 30+ 种数据源。我在项目中用它做跨集群的数据迁移，稳定性比 Sqoop 好不少。
DataX 的核心是「Reader + Channel + Writer」的三段式架构。你只需要写一个 JSON 配置文件，定义好从哪里读、怎么写、并发多少。
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"connection": [
{ "jdbcUrl": ["jdbc:mysql://192.168.1.100:3306/biz_db"],
"table": ["user_profile"] }
],
"column": ["user_id", "name", "age", "update_time"],
"splitPk": "user_id",
"where": "update_time >= '2024-01-01'"
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"path": "/data/dwd/user_profile",
"fileName": "user_profile",
"writeMode": "append",
"fileType": "orc"
}
}
}
],
"setting": {
"speed": { "channel": 8 }
}
}
}你想想看，DataX 的 splitPk 参数和 Sqoop 的 split-by 是一个道理，都是用来做数据分片的。但 DataX 的 channel 控制更精细，可以单独限制每个 channel 的流量，避免把源库打崩。
3.2 流式采集：Kafka 与 Flink
批式采集适合「昨天的事」，但模型迭代越来越需要「现在的事」。比如实时推荐、风控拦截、在线学习，这些场景必须用流式采集。
3.2.1 Kafka：消息队列的基石
Kafka 是流式采集的「高速公路」。所有实时数据先往 Kafka 里扔，下游再各取所需。我习惯把 Kafka 的 topic 设计成「数据源 + 业务域」的命名方式，比如 ods_user_behavior_click、ods_order_paid。
生产端要注意消息的 key 设计。如果 key 不均匀，会导致某些 partition 数据倾斜。我曾经遇到过一个 topic 的 12 个 partition 里，有 3 个 partition 的数据量是其他 9 个的 10 倍，结果消费端 lag 越来越大。后来把 key 改成 user_id % partition_num 才解决。
3.2.2 Flink：真正的流处理引擎
Flink 是目前流处理的事实标准。它支持 exactly-once 语义、事件时间处理、状态管理，这些特性让实时 ETL 变得可靠。我个人习惯用 Flink SQL 做简单的清洗和转换，复杂逻辑才用 DataStream API。
-- Flink SQL 实时清洗用户行为数据
CREATE TABLE source_kafka (
user_id BIGINT,
event_type STRING,
event_time TIMESTAMP(3),
page_url STRING,
proc_time AS PROCTIME()
) WITH (
'connector' = 'kafka',
'topic' = 'ods_user_behavior',
'properties.bootstrap.servers' = 'kafka:9092',
'properties.group.id' = 'flink_etl_group',
'format' = 'json',
'scan.startup.mode' = 'latest-offset'
);
CREATE TABLE sink_hbase (
rowkey STRING,
cf ROW<event_type STRING, event_time STRING, page_url STRING>,
PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
'connector' = 'hbase-2.2',
'table-name' = 'dwd:user_behavior',
'zookeeper.quorum' = 'zk1:2181,zk2:2181'
);
INSERT INTO sink_hbase
SELECT
CAST(user_id AS STRING) || '_' || DATE_FORMAT(event_time, 'yyyyMMddHHmmss'),
ROW(event_type, CAST(event_time AS STRING), page_url)
FROM source_kafka
WHERE event_type IS NOT NULL;嗯，这里要注意：Flink 的 checkpoint 间隔不要设得太短。我见过有人设成 100ms，结果状态后端频繁快照，把 HDFS 的 NameNode 搞挂了。建议生产环境设 1~3 分钟，同时开启增量 checkpoint。
3.3 数据源适配器设计
不管是批式还是流式，你都会遇到同一个问题：数据源种类太多。MySQL、Oracle、PostgreSQL、MongoDB、Elasticsearch、API 接口……每个数据源的连接方式、认证机制、数据类型都不一样。如果每个数据源都写一套采集逻辑，代码会变得又臭又长。
所以，我们需要一个「数据源适配器」来统一管理。说白了，就是定义一个抽象接口，让每种数据源自己去实现连接、读取、写入的逻辑。
3.3.1 适配器接口设计
我习惯用 Java 的接口 + 抽象类来做。核心接口就三个方法：connect()、read()、close()。
public interface DataSourceAdapter {
// 建立连接
boolean connect(DataSourceConfig config);
// 读取数据，返回迭代器
Iterator<DataRecord> read(ReadContext context);
// 关闭连接
void close();
}
public abstract class BaseDataSourceAdapter implements DataSourceAdapter {
protected DataSourceConfig config;
protected Connection connection;
@Override
public boolean connect(DataSourceConfig config) {
this.config = config;
// 子类实现具体的连接逻辑
return doConnect();
}
protected abstract boolean doConnect();
@Override
public void close() {
if (connection != null) {
connection.close();
}
}
}具体的数据源只需要继承 BaseDataSourceAdapter，实现 doConnect() 和 read() 就行。比如 MySQL 适配器：
public class MysqlAdapter extends BaseDataSourceAdapter {
@Override
protected boolean doConnect() {
this.connection = DriverManager.getConnection(
config.getJdbcUrl(),
config.getUsername(),
config.getPassword()
);
return connection != null;
}
@Override
public Iterator<DataRecord> read(ReadContext context) {
String sql = buildSelectSql(context);
// 执行查询并返回迭代器
return new ResultSetIterator(connection.createStatement().executeQuery(sql));
}
}3.3.2 适配器注册与工厂模式
有了适配器，还得有个地方管理它们。我一般用工厂模式 + SPI 机制，让每种数据源自动注册。
public class DataSourceAdapterFactory {
private static final Map<String, Class<? extends DataSourceAdapter>> adapterMap = new HashMap<>();
static {
// 注册内置适配器
register("mysql", MysqlAdapter.class);
register("oracle", OracleAdapter.class);
register("kafka", KafkaStreamAdapter.class);
register("http", HttpApiAdapter.class);
}
public static void register(String type, Class<? extends DataSourceAdapter> clazz) {
adapterMap.put(type.toLowerCase(), clazz);
}
public static DataSourceAdapter create(String type, DataSourceConfig config) {
Class<? extends DataSourceAdapter> clazz = adapterMap.get(type.toLowerCase());
if (clazz == null) {
throw new UnsupportedDataSourceException("Unsupported data source type: " + type);
}
DataSourceAdapter adapter = clazz.getDeclaredConstructor().newInstance();
adapter.connect(config);
return adapter;
}
}这样做的好处是：新增数据源时，只需要写一个适配器类，然后在 static 块里注册一下就行。采集框架的代码完全不用改。
connect()、read()、close() 接口。这样，模型迭代时换数据源就像换插头一样简单。
3.4 本章知识体系总览
下面这张图把批式采集、流式采集和适配器设计的关系串起来了。你可以把它当作本章的「地图」。
3.5 小结
这一章我们聊了三个核心点：
- 批式采集：Sqoop 适合传统 Hadoop 生态，DataX 适合异构数据源。注意控制并发，别把源库打崩。
- 流式采集：Kafka 做消息缓冲，Flink 做实时 ETL。Checkpoint 间隔别太短，key 设计要均匀。
- 数据源适配器：用接口 + 工厂模式统一管理，新增数据源只需写一个适配器类。