4、数据采集与接入：储能BMS数据采集协议（Modbus/CAN）、日志采集（Filebeat/Flume）、实时采集（Kafka）、全量/增量同步策略
数据采集是整个数据仓库的「水源」。水源不干净，后面再怎么折腾也是白搭。我在储能行业摸爬滚打这些年，见过太多项目因为采集层没设计好，导致后续分析全成了「垃圾进，垃圾出」。今天咱们就聊聊储能场景下，数据到底该怎么「接」进来。
核心观点：储能数据采集不是简单的「把数据拿过来」，而是要解决三个问题——怎么连、怎么传、怎么存。BMS协议、日志工具、消息队列、同步策略，缺一不可。
4.1 储能BMS数据采集协议：Modbus与CAN
BMS（电池管理系统）是储能系统的「大脑」。它监控着每一节电芯的电压、温度、电流、SOC（荷电状态）等关键参数。要拿到这些数据，你得先跟BMS「对话」。
Modbus协议，说白了就是工业界的「普通话」。它基于主从架构，你发请求，设备回数据。我习惯用Modbus TCP，走以太网，速度快、布线简单。但要注意，Modbus的寄存器地址各家厂商定义不一样，你得先拿到设备手册。
// 示例：用Python读取BMS的Modbus数据
from pymodbus.client import ModbusTcpClient
client = ModbusTcpClient('192.168.1.100', port=502)
client.connect()
# 读取保持寄存器，起始地址0x3100，读取10个寄存器
result = client.read_holding_registers(0x3100, 10, unit=1)
if not result.isError():
voltage = result.registers[0] / 1000.0 # 单位转换
current = result.registers[1] / 1000.0
soc = result.registers[2] / 10.0
print(f"总电压: {voltage}V, 电流: {current}A, SOC: {soc}%")
client.close()
CAN协议则更「硬核」。它常用于车载或分布式储能系统，抗干扰能力强。但CAN的数据是二进制帧，你得自己解析ID和报文格式。我曾经在某个项目中，BMS厂商给的CAN协议文档有20多页，光解析就花了两天。
我的经验：如果现场既有Modbus又有CAN，我建议优先用Modbus。CAN的调试成本高，而且数据量一大就容易丢帧。除非是车规级场景，否则别给自己找麻烦。
4.2 日志采集：Filebeat vs Flume
BMS数据是结构化的，但储能系统还有大量非结构化日志——比如逆变器告警、通信异常、运维操作记录。这些日志怎么收？
Filebeat，轻量级选手。它基于Go语言开发，资源占用极低。我习惯在每台储能柜的工控机上部署一个Filebeat，直接读取日志文件，然后发到Kafka或Elasticsearch。配置起来也简单：
# filebeat.yml 配置示例
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/bms/*.log
- /var/log/inverter/*.log
output.kafka:
hosts: ["kafka1:9092", "kafka2:9092"]
topic: "storage-logs"
partition.round_robin:
reachable_only: true
Flume，重量级选手。它基于Java，功能更强大，支持复杂的拦截器、通道选择器、Sink处理器。但说实话，在储能场景下，Flume有点「杀鸡用牛刀」。除非你的日志量特别大（比如每秒几万条），否则Filebeat完全够用。
避坑指南：我曾经在一个项目中用Flume采集日志，结果因为Java堆内存配置不当，导致工控机频繁OOM。后来换成Filebeat，CPU占用从30%降到了5%。记住：储能工控机的资源很宝贵，能轻则轻。
4.3 实时采集：Kafka
Kafka是实时数据流的「高速公路」。BMS的秒级数据、逆变器的实时功率、环境传感器的温湿度……这些数据都需要先进入Kafka，再由下游的Flink或Spark Streaming消费。
我设计Kafka集群时，一般遵循几个原则：
- 分区数 = 消费者数 × 2，保证并行度
- 副本因子 = 3，防止数据丢失
- 消息保留7天，给下游重跑留足时间
举个例子，BMS每秒钟上报一次数据，一个储能站有100台BMS，每台上报100个指标。算下来每秒1万条消息，Kafka轻松扛住。
// 生产者示例：将BMS数据发送到Kafka
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer producer = new KafkaProducer<>(props);
String msg = "{\"bms_id\":\"BMS-001\",\"voltage\":\"48.5\",\"current\":\"120.3\",\"soc\":\"85.2\"}";
producer.send(new ProducerRecord<>("bms-raw-data", "BMS-001", msg));
producer.close();
小技巧：Kafka的Topic命名我建议用「数据源-数据类型-时间粒度」的格式，比如bms-raw-1s、inverter-alarm-5s。这样后续维护时，一眼就能看出这个Topic是干嘛的。
4.4 全量/增量同步策略
数据仓库里不光有实时数据，还有历史数据。比如你要把过去一年的BMS数据从业务库同步到数仓，这就涉及到同步策略。
全量同步，简单粗暴。把源表的所有数据一次性拉过来。适合数据量小、变更不频繁的场景。比如储能设备的型号表、厂商信息表，一个月同步一次就行。
增量同步，细水长流。只同步新增或变更的数据。常用的方式有：
- 时间戳字段：源表有
update_time，每次同步取大于上次最大时间的数据 - Binlog监听：通过Canal或Debezium监听MySQL的binlog，实时捕获变更
- 日志对比：用MaxCompute或Spark对比两张表的差异，只同步变化部分
我个人的习惯是：首次全量，后续增量。第一次建仓时，跑一次全量同步把历史数据拉过来。之后每天跑增量同步，只处理当天变化的数据。这样既保证了数据完整性，又节省了计算资源。
| 同步方式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 全量同步 | 小表、低频变更 | 实现简单，数据完整 | 耗时长，占用资源多 |
| 增量同步（时间戳） | 有更新时间字段的表 | 效率高，资源消耗低 | 依赖时间戳准确性 |
| 增量同步（Binlog） | 实时性要求高的场景 | 毫秒级延迟，不丢数据 | 需要额外组件，运维复杂 |
注意：增量同步最怕「数据乱序」。比如一条记录先被更新，后被删除，如果同步顺序乱了，就会导致数据不一致。我建议在增量同步时加上event_time字段，按时间排序后再写入数仓。
知识体系总览
下面这张图，把本章的核心逻辑串起来了。从BMS协议到日志采集，再到Kafka和同步策略，每一步都是环环相扣的。
你看，从数据源到存储层，每一层都有对应的技术和策略。Modbus/CAN解决「怎么连」，Filebeat/Flume解决「怎么收」，Kafka解决「怎么传」，全量/增量解决「怎么同步」。把这四块搞明白了，数据采集这块基本就稳了。
最后说一句：数据采集没有银弹。每个储能站的环境都不一样，有的用Modbus，有的用CAN，有的日志格式乱七八糟。我的建议是——先摸清现场，再定方案。别一上来就上Kafka集群，可能一个Filebeat就搞定了。