2. 实验数据采集与接入

实验数据批流一体处理的第一步，是把分散在不同来源、不同协议、不同格式的数据可靠地汇聚到统一通道中。这一步决定了后续处理的上限：如果采集阶段丢失了时间戳精度、单位信息或设备标识，后续再复杂的计算也无法还原真实实验过程。本节围绕四类典型数据源——传感器、日志、数据库、文件——说明各自的采集方法，并给出实时与批量接入协议的选择依据，最后落到数据格式标准化的具体做法。

四类数据源的采集方法

传感器数据通常由采集硬件或边缘网关产生，特点是高频、连续、带时间戳。常见接入方式有三种：一是通过 Modbus TCP/RTU 轮询寄存器，适合 PLC 与工业仪表；二是通过 MQTT 发布订阅，适合无线传感器网络，设备以 topic 区分测点；三是通过 OPC UA 订阅，适合需要语义化节点描述的产线设备。以 MQTT 为例，一个温度测点可发布到 lab/exp01/rig03/temp，载荷为 JSON：{"ts":1718000000123,"v":23.47,"unit":"C"}。采集端需要关注采样周期与时间戳来源：若设备本身带 RTC，应优先使用设备时间并记录时钟偏差；若设备无时钟，则由网关在接收时刻打点，并在元数据中标注 ts_source=gateway。

日志数据来自实验软件、仪器控制程序和操作系统，通常以文本行追加写入。采集方式包括文件尾部跟踪（如 tail -F 配合采集代理）、syslog 转发、以及应用内嵌 SDK 直接上报。日志的关键在于解析规则要稳定：建议在采集侧就完成一次结构化，把非结构化行转成带字段的记录，避免下游反复解析。例如一行 2024-06-10 09:12:33 INFO sample=17 voltage=3.301 可解析为 {ts, level, sample, voltage}。对于多行堆栈日志，需要配置行合并规则，否则会拆散一条完整事件。

数据库数据分为两类：实验业务库（如 MySQL、PostgreSQL）保存样品、批次、人员等元数据；时序库（如 InfluxDB、TimescaleDB）保存测量序列。批量接入常用 JDBC/ODBC 按时间窗口查询，实时接入则依赖 CDC（变更数据捕获），如 Debezium 读取 binlog 或 WAL，把 insert/update 事件转为流。CDC 的价值在于不侵入业务库，但要注意事务边界：一个事务的多条变更应保持同一批次标识，否则下游会看到中间状态。

文件数据包括仪器导出的 CSV、二进制原始文件、图像和视频。采集方式有目录监听（inotify/轮询）、对象存储事件通知、以及 FTP/SFTP 投递。文件接入必须解决“文件是否写完”的问题：常见做法是要求写入方先写临时名再重命名，或约定 .done 标志文件；采集端只处理已完成的文件，避免读到半截内容。

实时与批量接入协议的选择

实时接入与批量接入不是互斥的，而是按数据时效需求分工。传感器和运行日志通常走实时通道，因为异常需要秒级发现；数据库历史数据和仪器导出文件通常走批量通道，因为数据量大、时效要求低。选择协议时可参考三个维度：延迟要求、吞吐量、以及是否需要重放。

实时通道推荐以 Kafka 作为统一缓冲层，采集端作为生产者，下游批流任务作为消费者。Kafka 的分区键应选择设备 ID 或实验 ID，保证同一测点的数据有序。对于 MQTT 设备，可用 MQTT-Kafka 桥接，把 topic 映射为 Kafka topic，并保留原始 QoS 等级。批量通道可用定时任务按小时或按实验批次拉取，写入同一 Kafka topic 或直接落到对象存储，再由批处理任务读取。这样批流两条路径最终汇入同一份标准化数据，避免“两套数据对不上”。

需要特别注意的是接入的幂等性。网络抖动或任务重试会导致重复数据，标准化记录中应包含唯一键，例如 source + metric + ts，下游按此去重。对于文件接入，可用文件内容哈希作为幂等键；对于 CDC，可用数据库事务 ID 加行主键。

数据格式标准化的落地做法

标准化不是简单地把字段改名，而是建立一套所有来源都必须遵守的记录契约。建议统一为如下 JSON 结构，批流共用：

{
  "ts": 1718000000123,
  "source": "rig03.temp",
  "metric": "temperature",
  "value": 23.47,
  "unit": "C",
  "quality": "good",
  "batch_id": "exp01-20240610",
  "ts_source": "device"
}

其中 ts 统一为 UTC 毫秒；unit 必须显式给出，温度统一为摄氏度，压力统一为 kPa，电压统一为 V；quality 用于标记异常值，例如超量程记为 suspect，通信中断补零记为 bad。标准化应在采集侧或接入侧尽早完成，越早统一，下游解析成本越低。

对于 CSV 文件，可在采集时按列映射表转换：把 Time, Temp(C), Volt(V) 映射为 ts, temperature, voltage，并补上 source 与 batch_id。对于日志，用正则提取字段后同样套用该结构。对于数据库，查询时用 SQL 别名对齐字段名，避免下游再做映射。

一个可复用的接入检查清单

在实际搭建采集链路时，可按以下清单逐项确认，减少返工：

• 每个数据源是否已明确时间戳来源与精度？
• 传感器是否携带单位？日志解析规则是否覆盖多行与异常行？
• 数据库批量查询是否按时间窗口且可重跑？CDC 是否保持事务边界？
• 文件是否先判完成再读取？是否有幂等键防止重复？
• 实时与批量是否写入同一标准化通道？
• 标准化记录是否包含 ts、source、metric、value、unit、quality、batch_id？
• 是否配置了采集延迟、丢失率、解析失败率的监控？

完成上述工作后，采集与接入阶段就为批流一体处理提供了干净、统一、可追溯的输入。后续的窗口计算、状态管理和结果输出，都建立在这份标准化记录之上。