第3章数据采集与接入

3、数据采集与接入：多源异构数据源

数据采集，说白了就是给数据中台「喂饭」。

储能系统里，数据源五花八门。BMS、PCS、EMS、气象站……每个系统都有自己的脾气。有的吐结构化数据，有的吐半结构化日志，有的干脆就是个二进制流。我刚开始做储能项目时，光对接这些数据源就折腾了两周。

嗯，今天我们就聊聊怎么把这些「刺头」数据统一接入中台。

3.1 多源异构数据源全景

先看一张图，帮你快速建立全局认知。

这张图展示了从数据源到中台的全链路。你会发现，采集层是承上启下的关键节点。

3.2 实时流采集：Kafka 是主角

储能场景下，实时数据是命脉。BMS上报的电压、温度、SOC，PCS的功率、频率，EMS的调度指令——这些数据延迟超过1秒，就可能影响安全控制。

为什么选Kafka？

高吞吐：单机每秒能处理几十万条消息。我见过一个100MW的储能站，每秒产生约5万条测点数据，Kafka轻松扛住。
持久化：数据落盘，支持回溯。有一次现场排查故障，我们回放了3小时前的数据流，很快就定位到了问题。
解耦：生产者和消费者互不依赖。BMS只管往Topic里写，下游的实时计算、离线分析各取所需。

核心架构要点：

每个数据源对应一个Topic。比如：topic_bms_raw、topic_pcs_raw、topic_ems_raw。

分区数建议设置为3~6个，兼顾并行度和运维成本。

我习惯在采集端加一层协议适配。BMS走Modbus TCP，PCS走IEC 104，EMS走OPC UA。统一转换成JSON格式后，再写入Kafka。这样下游消费时就不用关心底层协议了。

// 伪代码：BMS数据采集适配器
public class BmsCollector {
    public void collect() {
        // 1. 连接BMS（Modbus TCP）
        ModbusClient client = new ModbusClient("192.168.1.100", 502);
        // 2. 读取寄存器
        float voltage = client.readFloat(0x1000);
        float current = client.readFloat(0x1002);
        float soc = client.readFloat(0x1004);
        // 3. 封装为JSON
        String message = String.format(
            "{\"timestamp\":%d,\"voltage\":%.2f,\"current\":%.2f,\"soc\":%.2f}",
            System.currentTimeMillis(), voltage, current, soc
        );
        // 4. 发送到Kafka
        kafkaProducer.send("topic_bms_raw", message);
    }
}

避坑指南：

我曾经遇到过BMS频繁断连的问题。后来发现是Modbus的超时设置太短。建议把超时时间设为5秒，重试次数3次。另外，别忘了加心跳检测，每30秒发一次。

3.3 批量采集：Sqoop 与 DataX

实时数据搞定了，那历史数据怎么办？

储能系统里，很多数据是批量产生的。比如：

EMS的日发电量报表
BMS的月度健康度评估数据
气象站的历史辐照度记录

这些数据存在关系型数据库里，需要定期同步到数据中台。

Sqoop vs DataX：怎么选？

特性	Sqoop	DataX
数据源	主要面向Hadoop生态	支持MySQL、Oracle、HDFS、Hive等20+种
性能	MapReduce模式，适合大表	多线程并发，小表更灵活
运维	依赖Hadoop集群	独立部署，轻量级
增量同步	支持（--incremental参数）	支持（where条件过滤）

我个人习惯：大表用Sqoop，小表用DataX。

举个例子：BMS的月度健康数据表有上亿行，用Sqoop的MapReduce并行读取，20分钟搞定。而EMS的日发电量报表只有几千行，用DataX的线程池模式，2分钟就同步完了。

# Sqoop 增量同步示例
sqoop import \
  --connect jdbc:mysql://192.168.1.200:3306/ems_db \
  --username ems_user \
  --password ems_pass \
  --table daily_report \
  --incremental lastmodified \
  --check-column update_time \
  --last-value '2024-01-01 00:00:00' \
  --target-dir /data/ems/daily_report \
  --num-mappers 4

注意：

批量采集的时间窗口要避开业务高峰期。储能站一般在凌晨2点到5点进行数据同步，这时候充放电任务少，系统负载低。

我曾经有一次把同步任务设在上午10点，结果把BMS的数据库连接池打满了，导致实时数据采集中断了3分钟。嗯，从那以后我再也不敢在高峰期跑批量任务了。

3.4 采集链路监控

数据采集链路长了，出问题的地方就多。我建议至少监控三个指标：

采集延迟：从数据产生到进入Kafka的时间差。超过5秒就要报警。
数据丢失率：对比源端和Kafka的消息数。丢失率超过0.1%就要排查。
采集节点健康度：CPU、内存、网络IO。特别是Modbus采集节点，CPU飙升往往是协议解析出了问题。

说白了，数据采集是数据中台的「第一公里」。这公里跑不顺，后面再好的算法、再炫的报表都是空中楼阁。