一、数据采集与接入：从源头说起

数据全链路管理的第一步，就是怎么把数据弄进来。你想想看，模型再牛，算法再先进，源头数据进不来或者进错了，后面全是白搭。我见过太多项目，模型精度上不去，最后排查下来，问题出在数据采集阶段——要么丢字段，要么时间戳对不上，要么格式乱成一锅粥。

所以这一章，咱们先把数据源类型、采集工具、接入策略这三个核心问题讲透。嗯，这里要注意，不是让你把所有工具都学一遍，而是理解它们各自的定位和适用场景。

1.1 数据源类型：你面对的是什么数据？

数据源按结构程度，可以分成三类。我个人习惯用「结构化程度」来划分，因为这会直接影响你选什么工具、用什么方式接入。

类型	特点	典型例子	接入难度
结构化数据	严格的行列格式，Schema 固定	MySQL、PostgreSQL、Oracle 表	低
半结构化数据	有自描述结构，但 Schema 灵活	JSON、XML、YAML、Avro	中
非结构化数据	无固定结构，通常是二进制或文本	图片、视频、日志文本、PDF	高

我在项目中遇到过最头疼的情况：业务方说「数据都在 Kafka 里，JSON 格式，很干净」。结果一接进来，同一个字段有时是字符串，有时是数字，有时干脆缺失。这就是半结构化数据的典型坑——Schema 不强制。

结构化数据

说白了就是关系型数据库里的表。行和列清清楚楚，每个字段的类型、长度、约束都定义好了。接入这类数据，你只需要关注两点：连接方式和同步频率。工具方面，Sqoop、DataX、Kettle 都能干这活。

半结构化数据

这是目前最主流的数据形态。JSON 和 Avro 在日志、事件、API 响应里随处可见。它们有标签、有层级，但同一个字段在不同记录里可能类型不一致。处理这类数据，我建议你提前做 Schema 推断和校验，否则下游解析时很容易报错。

非结构化数据

图片、视频、PDF、原始日志文本……这类数据没有固定结构，接入时通常先存成对象（比如 HDFS 或 S3 上的文件），然后在处理阶段再解析。我曾经踩过一个坑：直接把 PDF 当文本流接入，结果编码没处理好，中文全乱码了。后来学乖了，非结构化数据接入时，先确认编码和格式。

1.2 数据采集工具：选对工具，事半功倍

工具选型是个老生常谈的话题。我不打算给你列一堆对比表，而是说说我实际用下来的感受。

Flume

Flume 是 Hadoop 生态的老牌采集工具。它的核心设计是 Agent 架构——Source、Channel、Sink 三段式。Source 负责收数据，Channel 做缓冲，Sink 负责写出去。

# 一个简单的 Flume 配置示例
agent.sources = r1
agent.channels = c1
agent.sinks = k1

agent.sources.r1.type = spooldir
agent.sources.r1.spoolDir = /data/logs

agent.channels.c1.type = memory
agent.channels.c1.capacity = 10000

agent.sinks.k1.type = hdfs
agent.sinks.k1.hdfs.path = /user/flume/events/%Y%m%d

Flume 的优势是轻量、稳定，适合日志文件的实时采集。但它的短板也很明显：扩展性一般，数据可靠性依赖 Channel 配置。我建议你只在 Hadoop 生态内、数据量不太大的场景下用 Flume。

Logstash

Logstash 是 ELK 栈的核心组件。它的插件生态非常丰富，输入、过滤、输出各有几十种插件。我最喜欢它的 filter 阶段——可以实时解析、转换、清洗数据。

# Logstash 配置示例
input {
  file {
    path => "/var/log/nginx/access.log"
    start_position => "beginning"
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
  }
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
  }
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "nginx-access-%{+YYYY.MM.dd}"
  }
}

Logstash 的优点是灵活、易用，但性能是硬伤。数据量大时（比如每秒上万条），Logstash 很容易成为瓶颈。我有个项目，Logstash 实例跑在 8 核机器上，每秒处理 5000 条日志就 CPU 飙到 90%。后来换成了 Kafka + Flink 的方案。

Kafka

Kafka 严格来说不只是一个采集工具，它更是一个分布式消息队列。但在全链路管理中，Kafka 经常被用作数据采集的「中转枢纽」。生产者把数据发到 Topic，消费者从 Topic 拉取数据。

# 生产者示例（Java）
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key", "value"));

Kafka 的优势是高吞吐、持久化、可回溯。我现在的项目，所有日志和事件数据都先进 Kafka，下游再用 Flink、Spark Streaming 或 Logstash 消费。这样做的好处是：解耦了数据生产和消费，生产端不用关心下游怎么处理。

工具	核心定位	吞吐量	适用场景
Flume	日志采集 + 传输	中等	Hadoop 生态、文件采集
Logstash	采集 + 过滤 + 输出	较低	ELK 栈、日志解析
Kafka	消息队列 + 缓冲	高	高吞吐、解耦、流处理

1.3 数据接入策略：实时还是批量？

这是个经典选择题。很多刚入行的同学觉得「实时就是好，批量就是落后」。其实不是这么回事。你想想看，实时接入意味着更高的成本、更复杂的运维、更严格的容错要求。而批量接入简单、稳定、成本低。

实时接入

实时接入的核心是低延迟。数据产生后，秒级甚至毫秒级就能被下游消费。典型场景包括：

• 风控系统：交易发生后立即判断是否异常
• 监控告警：系统指标异常时马上通知
• 实时推荐：用户行为发生后立即更新推荐结果

实时接入的技术栈通常是 Kafka + Flink/Spark Streaming。我做过一个实时风控项目，要求端到端延迟小于 500 毫秒。当时用了 Kafka 做缓冲，Flink 做计算，Redis 做特征存储。嗯，这里要注意，实时不等于「零延迟」，而是「可接受的延迟」。

批量接入

批量接入的核心是高吞吐和低成本。数据按固定时间窗口（比如每小时、每天）统一拉取或推送。典型场景包括：

• 离线数仓：每天凌晨同步前一天的业务数据
• 模型训练：每周更新一次训练数据集
• 报表统计：每月生成一次财务报表

批量接入的工具很多：Sqoop、DataX、Kettle、甚至简单的 Shell 脚本 + crontab。我建议你能用批量就别用实时，除非业务真的需要秒级响应。

维度	实时接入	批量接入
延迟	秒级/毫秒级	分钟级/小时级
吞吐量	受限于处理能力	可批量处理，吞吐高
成本	高（计算、存储、运维）	低
复杂度	高（状态管理、容错、Exactly-Once）	低（重试、断点续传）
适用场景	风控、监控、推荐	数仓、报表、模型训练

1.4 知识体系总览

下面这张图把本章的核心知识点串起来了。你可以看到，数据源类型决定了采集工具的选择，采集工具的输出又决定了接入策略。三者环环相扣。

从这张图可以看得很清楚：数据源类型决定了你用什么工具，工具的输出决定了你用什么策略，策略又决定了数据最终流向哪里。没有的最优方案，只有最适合你业务场景的方案。