📚 课程目录第4章标签数据源接入

4、标签数据源接入：结构化/半结构化/非结构化数据，实时与批量采集方案

数据源接入，是整个标签策略自动化的第一步，也是最容易翻车的一步。

我刚开始做标签系统的时候，总觉得数据源嘛，无非就是配个连接、拉个数据。结果呢？线下跑得好好的，上线第一天就把 Kafka 集群搞挂了。为什么？因为没弄清楚数据源到底长什么样，也没想清楚该用批采还是实时采。

这一章，我就把三种数据源的识别方法和采集方案一次性讲透。你看完就能对着自己的系统直接套用。

4.1 三类数据源，识别清楚再动手

先别急着写代码。你得先判断：手里的数据源，到底是结构化、半结构化，还是非结构化？判断错了，后面的采集方案就是白搭。

我习惯用一个简单粗暴的方法：看「结构容忍度」。

数据类型	典型代表	结构容忍度	我建议的采集方式
结构化	MySQL、Oracle、Hive 表	高（必须强 Schema）	JDBC/ODBC 批量同步
半结构化	JSON、XML、日志文件	中（Schema 可有可无）	文件监听 + 流式解析
非结构化	图片、视频、PDF、音频	低（无固定结构）	对象存储 + 元数据索引

结构化数据，说白了就是每行都有固定的列和类型。你在 MySQL 里建表的 CREATE TABLE 写得越细致，它就越结构化。接入这类数据，我最怕的是「字段膨胀」——有一次业务方临时加了个列，没通知我，批量采集直接崩了。后来我加了个 Schema 校验层，采集前先对比元数据，新字段自动告警，不盲采。

半结构化数据，你想想看，JSON 它允许 key 随意增减，字段层级也能变。接入这类数据，我的策略是「先把结构摊平」——不管多深的嵌套，转换成宽表再做标签加工。曾经有个项目，业务日志里嵌套了 5 层 JSON，直接按原始结构做标签抽取，性能惨不忍睹。后来我改成预解析成 Parquet，效率提升了 10 倍。

非结构化数据，比如用户上传的身份证照片、客服聊天录音。这类数据本身不能直接产标签，但我们能从它的「元数据」里提炼特征：文件大小、上传时间、OCR 结果、语音转文字后的关键词。嗯，这里要注意：元数据采集一定要做「增量」，否则每次全量扫描对象存储，成本真的扛不住。

避坑指南：非结构化数据不要直接接入标签计算引擎。先把元数据抽出来，存入 OLAP 表，然后再参与标签逻辑。我以前直接拿图片 URL 做标签入参，结果标签系统跑了一下午，IO 全打满了。

4.2 批量采集与实时采集，数据源决定方案

搞清楚了数据源长什么样，接下来要选采集模式。我的原则就一句话：业务容忍多长时间的延迟，就选多慢的采集。

4.2.1 批量采集：稳定、可控、适合 T+1

批量采集适合那些不需要秒级刷新的标签数据源。比如用户的基础信息（性别、注册时间），这些字段通常一天改一次就算频繁了。

我常用的批量采集框架是 DataX 和 Sqoop。用起 DataX 来，真的要夸一下它的 Json 配置结构，清晰得不像国产轮子。来看一个我从生产环境扒下来的简化配置：

{
    "job": {
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "connection": [
                            {
                                "jdbcUrl": ["jdbc:mysql://标签源库:3306/user_base"],
                                "table": ["user_info"]
                            }
                        ],
                        "splitPk": "user_id",
                        "column": ["user_id", "gender", "birthday", "city"]
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "defaultFS": "hdfs://标签集群:8020",
                        "path": "/warehouse/ods/user_info/dt=20250412",
                        "writeMode": "overwrite",
                        "column": [
                            {"name": "user_id", "type": "bigint"},
                            {"name": "gender", "type": "string"},
                            {"name": "birthday", "type": "date"},
                            {"name": "city", "type": "string"}
                        ]
                    }
                }
            }
        ],
        "setting": {
            "speed": {"channel": 8}
        }
    }
}

这里有个细节：splitPk一定要用数字型主键，别用字符串。我曾经试过用用户手机号做拆分键，结果 DataX 直接按字典序分片，数据严重倾斜，8 个 channel 有 6 个闲着。

4.2.2 实时采集：低延迟、高吞吐、适合事件型标签

实时采集的数据源，典型的就是业务日志（埋点、点击流）、Kafka 消息队列。这类数据产生的标签大多跟「行为」有关：用户是否在 5 分钟内加购、是否刚完成支付。

我个人习惯首选 Canal 接入 MySQL binlog 做实时同步。CDC（Change Data Capture）方案的好处是：对源库无侵入，只读 binlog 不锁表。但有个坑：如果数据库做了主从切换，binlog 偏移量可能不连续，你得配合 ZK 记录位点。

# canal 简易配置（片段）
canal.instance.master.address = 192.168.1.100:3306
canal.instance.dbUsername = canal_user
canal.instance.dbPassword = canal_pass
canal.instance.connectionCharset = UTF-8
canal.instance.tsdb.enable = true
canal.instance.filter.regex = user_db\\..*
# 只监听 user_db 库的所有表
canal.mq.topic = user-binlog-topic
canal.mq.partitionsNum = 6

实时采集写进 Kafka 之后，标签引擎用 Flink 或者 Spark Streaming 接消费。我建议消费端一定要做「幂等写入」——万一任务重启，重复的数据不会产生两条相同的标签。

4.2.3 流批一体，能不能在一个框架里搞定？

很多团队都问过我这个问题。我的回答是：能用，但有代价。

Flink 从 1.12 开始就推流批一体了。如果你的整个数据链路都是 Flink 系（SQL 表、Hive Catalog、Kafka 表），那确实可以一套代码跑批和流。但如果你对接的源是传统 Oracle、Excel 文件，那还是老老实实分开吧——硬凑只会让你的代码变得又臭又长。

我自己的经验：80% 的场景批量采集就够了，只有那 20% 核心行为标签才值得走实时。别为了赶时髦把所有数据都接实时，不然你会被 Kafka 的运维成本和数据质量反哺搞疯掉。

实用建议：标签数据源接入后，一定要留一个「数据质量看板」。每天凌晨自动跑一次：采集行数、空值率、主键重复数、时间戳断层。发现问题立刻告警，别等业务发现标签不准了才去排查。

4.3 三种采集方案的选型对比

我把三种主流采集方案列了个表，方便你对照自己的场景选型：

方案	技术栈	延迟	典型场景	需要关注的点
离线批采	DataX / Sqoop / Spark	T+1 / 小时级	用户基础表、订单快照	数据倾斜、接口限流
实时流采	Canal / Debezium + Kafka + Flink	秒级	埋点事件、支付状态变化	位点丢失、重复消费
微批采集	Spark Streaming（批次时间 30s-5min）	分钟级	日志汇总、行为缓动	窗口乱序、watermark

选型的时候，我建议你带着这三个问题去问业务方：

这个标签，假设晚 10 分钟才更新，会影响线上业务判断吗？
如果某一批数据没采集进来，是「短暂缺失」可以接受，还是必须「精确一次」？
数据源的 QPS 峰值是多少？采集侧的连接数需要预留多少冗余？

把这三个问题聊清楚了，采集方案几乎就自动浮出水面了。

    核心认知：标签数据源的接入，不是简单的「读数据」，而是一次「结构认知 + 时效权衡 + 质量兜底」的系统工程。我踩过很多坑才总结出这个铁三角。只要在架构设计阶段就把这三件事定好，后面标签策略自动化落地才有真正的数据底气。

这张图是我每次做标签数据源架构设计时必画的脑图。从数据层到采集判定，再到处理层和输出层，每层之间都有一个明确的「决策点」。你以后设计采集方案，直接拿这张图对着填就行了。