4. Kafka Connect数据接入实战

机器学习平台的原始数据通常分散在三类系统中：业务数据库保存结构化样本与标签，对象存储保存图片、音频、Parquet 特征快照等大文件，日志系统保存行为埋点与在线服务调用记录。Kafka Connect 的价值在于用统一的 Source Connector 把这些异构数据持续写入 Kafka 主题，让下游 Flink、Spark Structured Streaming 或特征工程服务只面对 Kafka 这一种输入接口。本节围绕三类数据源，讲清连接器选型、关键参数、数据落主题后的形态，以及如何判断同步是否正常。

一、数据库接入：Debezium CDC 与 JDBC 增量拉取

数据库接入有两种典型方式。第一种是 Debezium 系列 CDC 连接器，通过读取 MySQL binlog、PostgreSQL WAL 捕获行级变更，把 INSERT/UPDATE/DELETE 转成带 before、after、op 字段的变更事件。第二种是 JDBC Source Connector 的增量模式，依靠自增主键或时间戳列周期性查询。对机器学习样本表，如果存在硬删除或需要精确还原历史状态，应优先选 CDC；如果只是周期性拉取新增样本，JDBC 增量更简单。

以 MySQL 为例，先确认 binlog 格式为 ROW，并创建具备 REPLICATION SLAVE、REPLICATION CLIENT 权限的账号。随后提交 Debezium MySQL 连接器配置：

{
  "name": "mysql-ml-samples-cdc",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-ml.internal",
    "database.port": "3306",
    "database.user": "kafka_cdc",
    "database.password": "******",
    "database.server.id": "5401",
    "topic.prefix": "mlcdc",
    "database.include.list": "ml_platform",
    "table.include.list": "ml_platform.training_sample,ml_platform.sample_label",
    "snapshot.mode": "initial",
    "decimal.handling.mode": "double",
    "time.precision.mode": "connect",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

这里 snapshot.mode=initial 表示首次启动先做全量快照，再切换到 binlog 增量；ExtractNewRecordState 把 Debezium 的复杂信封压平成业务字段，便于下游直接解析。decimal.handling.mode=double 是为了让金额、评分等字段以数值而非 Base64 出现，避免特征工程侧再做解码。连接器启动后，每个表会落到 mlcdc.ml_platform.training_sample 这类主题，分区数建议与表更新并发度匹配，通常设为 6～12。

如果使用 JDBC 增量模式，核心参数是 mode=incrementing、incrementing.column.name=id、poll.interval.ms=5000。它无法捕获 UPDATE 和 DELETE，因此只适合 append-only 的样本表。判断同步是否正常，可以观察连接器状态是否为 RUNNING、主题 end offset 是否持续增长，以及抽样消费一条消息确认字段完整。

二、对象存储接入：S3 事件驱动与定时清单扫描

对象存储中的机器学习原始数据多为图片、音频、TFRecord 或 Parquet 分片。S3 Source Connector 支持两种发现方式：一是基于 SQS 事件通知，文件一落地就触发；二是基于 s3.bucket.name 加前缀的定时轮询。前者延迟低，适合实时样本回流；后者实现简单，适合批量历史数据导入。

关键配置包括 s3.bucket.name、s3.prefix、topics.dir 或 topic、file.data.field。默认情况下连接器会把对象内容以字节数组写入消息，若文件较大，应设置 max.partition.records 或改用 ByteArrayConverter 并配合下游解析。对于 Parquet，更常见的做法是只同步对象路径和元数据，由下游 Spark 按路径读取，避免把大文件塞进 Kafka。

{
  "name": "s3-ml-images-source",
  "config": {
    "connector.class": "io.confluent.connect.s3.source.S3SourceConnector",
    "tasks.max": "4",
    "s3.bucket.name": "ml-raw-images",
    "s3.prefix": "daily/2025/",
    "s3.region": "ap-east-1",
    "format.class": "io.confluent.connect.s3.format.bytearray.ByteArrayFormat",
    "mode": "GENERIC",
    "topic": "ml.raw.image.meta",
    "file.data.field": "image_bytes",
    "behavior.on.error": "fail"
  }
}

实际生产中更推荐“元数据主题 + 对象路径”模式：连接器只把 s3://ml-raw-images/daily/2025/06/01/img_000123.jpg、文件大小、ETag、事件时间写入 ml.raw.image.meta，下游消费后再按需拉取对象。这样 Kafka 消息体积可控，也便于重放。判断同步是否正常，可对比 S3 对象数量与主题消息数量，并检查 s3.prefix 是否覆盖了所有分区路径。

三、日志系统接入：FileStream 与 Syslog 的取舍

日志类原始数据包括应用埋点、模型服务请求日志、容器 stdout。FileStream Source Connector 适合读取本地或挂载卷上的日志文件，按行写入主题；Syslog Source Connector 则直接监听 UDP/TCP 端口，适合网络设备或统一日志转发。两者都不做复杂解析，解析工作交给下游流处理。

FileStream 的核心参数是 file、topic、tasks.max。它按行读取，每行一条消息，因此要求日志本身是行分隔的 JSON 或文本。若日志文件会滚动，需要配合 file.rotate.interval.ms 或使用支持 tail 的替代连接器。Syslog 侧则关注 syslog.listener、syslog.port、topic，并注意 UDP 可能丢包，关键日志应走 TCP。

{
  "name": "filebeat-json-source",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "/var/log/ml-service/request.log",
    "topic": "ml.raw.service.log"
  }
}

日志接入后，主题中的消息是原始行文本。下游需要用 JsonConverter 或 Flink JSON 函数解析出 request_id、model_version、latency_ms、timestamp 等字段。判断同步是否正常，可以观察主题消息速率是否与日志写入速率一致，并抽样检查是否存在截断行或乱码。

四、三类数据源同步链路总览

下图把数据库、对象存储、日志系统三条接入链路放在同一张拓扑中，标出各自的连接器、关键参数、目标主题以及下游消费方式，便于对照选型与排障。

五、同步质量判断与常见问题

三类接入链路共用一套判断逻辑。第一看连接器状态，通过 GET /connectors/{name}/status 确认 state=RUNNING 且 tasks 无 FAILED。第二看主题偏移量，用 kafka-consumer-groups.sh --describe 或 Confluent Control Center 观察 end offset 是否随时间增长；若长期不变，说明源端无新数据或连接器卡住。第三做抽样消费，用 kafka-console-consumer.sh --topic mlcdc.ml_platform.training_sample --from-beginning --max-messages 3 检查字段是否完整、时间戳是否合理。

常见问题有三类。数据库侧最常见的是 binlog 权限不足或 server.id 冲突，表现为连接器反复重启；对象存储侧常见的是 s3.prefix 写错导致扫不到文件，或 IAM 缺少 s3:GetObject；日志侧常见的是文件滚动后连接器仍指向旧 inode，需要重启任务或改用支持 tail 的插件。对机器学习场景，还要额外关注时间字段：CDC 的 ts_ms、S3 对象的 LastModified、日志行内的时间戳必须统一到事件时间语义，否则下游窗口聚合会出现偏差。