数据源与采集：Kafka核心概念与架构、生产者与消费者API、Connect与CDC、Flume与Logstash对比

数据采集是整个流批一体架构的起点。说白了，数据进不来，后面再牛的计算引擎也是白搭。这一章我重点讲Kafka——实时数据管道的主角，再聊聊Flume和Logstash这两个老牌采集工具怎么选。

Kafka核心概念与架构

Kafka本质上是一个分布式消息系统。但它跟传统的消息队列（比如RabbitMQ）不太一样。我刚开始接触时也犯过迷糊——它更像一个日志系统，消息一旦写入就不会被消费删除，而是靠偏移量（offset）来标记消费位置。

几个核心概念你得先记住：

• Topic（主题）：消息的逻辑分类。一条订单数据发到"order_topic"，一条用户行为发到"user_behavior_topic"。
• Partition（分区）：Topic的物理分片。每个分区是一个有序的、不可变的日志文件。分区数决定了并行度。
• Broker（代理节点）：Kafka集群中的一台服务器。一个集群通常有3台以上Broker。
• Consumer Group（消费组）：一组消费者共同消费一个Topic。组内每个消费者负责不同的分区。
• Offset（偏移量）：消息在分区内的唯一序号。消费者通过提交offset来记录消费进度。

我画了一张架构图，帮你理清这些组件之间的关系：

Kafka生产者与消费者API

API这块其实不复杂，但坑不少。我一个个说。

生产者API

生产者的核心工作是发消息。但怎么发、发到哪个分区、失败了怎么办——这些才是关键。

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");
// acks=all 表示所有副本都确认才算成功，最安全但延迟最高
props.put("acks", "all");
// 重试次数
props.put("retries", 3);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// 异步发送 + 回调
producer.send(new ProducerRecord<>("my-topic", "key1", "value1"), 
    (metadata, exception) -> {
        if (exception != null) {
            System.err.println("发送失败: " + exception.getMessage());
        } else {
            System.out.println("发送成功，分区: " + metadata.partition() + 
                               "，偏移量: " + metadata.offset());
        }
    });

producer.close();

消费者API

消费者的核心是消费进度管理。你想想看，如果消费者挂了，重启后从哪里开始消费？这就是offset的作用。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 自动提交offset，默认5秒一次
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", 
                          record.offset(), record.key(), record.value());
    }
}

Kafka Connect与CDC技术

Kafka Connect是Kafka官方提供的数据集成框架。它的作用很简单——帮你把外部系统的数据搬进Kafka，或者从Kafka搬出去。不用写一行生产者和消费者代码。

CDC（Change Data Capture，变更数据捕获）是Connect最常用的场景之一。说白了，就是监听数据库的binlog（MySQL）或WAL（PostgreSQL），把增删改操作实时同步到Kafka。

组件	作用	常用场景
Source Connector	从外部系统读取数据到Kafka	MySQL CDC、文件采集、MQ消费
Sink Connector	从Kafka写入数据到外部系统	写入HDFS、Elasticsearch、HBase
Debezium	基于CDC的MySQL/PostgreSQL连接器	实时同步数据库变更到Kafka

举个例子，用Debezium监听MySQL的binlog：

# 启动一个MySQL Source Connector
curl -X POST http://localhost:8083/connectors -H "Content-Type: application/json" -d '{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "192.168.1.100",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.server.id": "1",
    "database.server.name": "my-server",
    "database.include.list": "mydb",
    "table.include.list": "mydb.orders",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.history.kafka.topic": "schema-changes.mydb"
  }
}'

启动之后，Kafka里会自动生成一个Topic叫my-server.mydb.orders，里面每条消息都对应orders表的一次变更。我做过一个实时数仓项目，就是用Debezium把MySQL的订单变更实时同步到Kafka，然后Flink消费做实时ETL，延迟不到1秒。

Flume与Logstash对比

这两个工具都是日志采集的老兵。但它们的定位和适用场景差别很大。我直接上对比表：

维度	Flume	Logstash
开发语言	Java	Ruby（JRuby）
核心定位	海量日志采集、聚合、传输	日志解析、过滤、转换、输出
数据源	Avro、Syslog、Spooling Directory、Kafka等	File、Syslog、TCP/UDP、Kafka、JDBC等
数据处理能力	弱，主要做传输，过滤能力有限	强，支持Grok、KV、Date等丰富filter插件
可靠性	高，支持Channel级别的持久化（File Channel）	中，默认内存队列，有丢数据风险
性能	高，单机可支撑数万条/秒	中，复杂filter会明显降低吞吐
配置方式	Java properties文件	YAML或JSON格式的pipeline配置
典型场景	大数据生态（Hadoop/Spark/Kafka）日志采集	ELK栈（Elasticsearch/Logstash/Kibana）日志处理

怎么选？我个人的经验是：

• 如果你在Hadoop/Spark生态里做日志采集，选Flume。它跟HDFS、Kafka的集成非常成熟，而且File Channel保证了数据不丢。我做过一个每天处理10TB日志的项目，Flume集群跑了两年没出过问题。
• 如果你在用ELK做日志分析，选Logstash。它的filter插件太强了，Grok正则解析、GeoIP地理定位、KV键值对提取，这些Flume做起来很费劲。
• 如果你既要高吞吐又要复杂处理，可以Flume + Logstash组合。Flume负责采集和传输，Logstash负责解析和输出到Elasticsearch。