2. 实验数据源与采集接入

实验数据湖仓一体平台的第一道工程关口，是把分散在实验室、产线、机房和业务系统中的数据“接进来”。接入做不好，后面的存储、建模、分析都会变成无源之水。本章围绕数据源识别、采集协议选择、实时与批量接入方法，以及一次可复现的接入任务实战展开。

2.1 先识别数据源，再谈采集

实验场景中的数据源通常分为四类。第一类是实验仪器，如示波器、频谱仪、电子万能试验机、液相色谱仪，它们往往通过 GPIB、RS-232、USB、LAN 或厂商私有接口输出测量结果，部分设备支持 SCPI 指令远程读取。第二类是传感器与采集卡，如温度、压力、振动、应变传感器，经 PLC、DAQ 或边缘网关汇聚，常见协议有 Modbus RTU/TCP、OPC UA、MQTT、CAN。第三类是日志与文件，包括设备运行日志、应用日志、实验软件导出的 CSV/JSON/Excel 文件，通常以文件或日志流形式产生。第四类是数据库与业务系统，如 LIMS、MES、ERP 中的结构化记录，需要通过 JDBC/ODBC 或 API 抽取。

识别数据源时，建议逐项记录五个要素：数据源名称与位置、数据形态（结构化/半结构化/非结构化）、产生频率与数据量、可用的访问接口、以及数据的时间语义（事件时间还是入库时间）。这五项决定了后续协议与接入方式的选择。例如，一台每秒输出 1000 个采样点的振动传感器，与每天生成一份检测报告的色谱仪，接入策略完全不同。

2.2 采集协议怎么选

协议选择的核心是匹配数据源的接口能力与实时性要求。对于传感器和 PLC，Modbus TCP 简单通用，适合周期性轮询寄存器；OPC UA 具备信息模型和订阅机制，适合需要语义化、变化上报的工业设备；MQTT 轻量、支持发布订阅，适合大量边缘设备经网关上报。对于仪器，若支持 SCPI，可用 Python 的 pyvisa 通过 TCP/IP 或 USB 发送指令读取；不支持远程接口的，则退化为文件导出加目录监听。对于数据库，优先使用 JDBC 增量抽取，按时间戳或自增主键拉取变更。

一个常见误区是“所有数据都走实时”。实际上，检测报告、日汇总、配置变更这类低频数据用批量接入更稳定、更易回溯；只有高频测量、设备状态、告警才需要实时通道。混合接入是常态。

2.3 实时接入与批量接入的实现要点

实时接入通常采用“采集端 → 消息队列 → 流处理 → 落湖”的链路。采集端将数据封装为带设备编码、采集时间、测点值、质量码的消息，发布到 Kafka 主题；流处理侧做窗口聚合与异常过滤，再以微批方式写入湖仓原始层。关键参数包括：批量大小（如 500 条或 2 秒触发一次）、重试次数、死信队列。若采集端网络中断，消息队列可起到缓冲作用，避免数据丢失。

批量接入以调度为核心。典型做法是用 Airflow 或 DolphinScheduler 定义每日/每小时任务：先从源库按 updated_at > 上次水位 抽取增量，落地为临时文件，再做字段映射与去重，最后写入 ODS 分区。水位值需持久化，防止重复抽取。对于文件类数据源，可用监听目录加文件指纹（大小+修改时间）判断是否为新文件。

2.4 接入任务实战：温度传感器与检测报告双源接入

假设某材料实验台有两类数据：一是 8 个温度传感器，经 Modbus TCP 网关每 5 秒上报一次；二是电子万能试验机每天导出一份 CSV 检测报告。目标是把两者接入湖仓 ODS 层。

实时侧：用 Python 编写采集脚本，通过 pymodbus 轮询网关寄存器，读取温度值并附加设备编码与时间戳，发布到 Kafka 主题 lab.sensor.temperature。示例配置如下：

# 采集端关键参数（示例）
gateway_host = "192.168.10.21"
gateway_port = 502
unit_id = 1
register_start = 0
register_count = 8
poll_interval = 5          # 秒
kafka_topic = "lab.sensor.temperature"
batch_size = 500           # 或满 2 秒触发
retry = 3
dlq_topic = "lab.sensor.temperature.dlq"

流处理侧消费该主题，按设备编码与分钟窗口计算平均温度，同时保留原始明细写入 ods.sensor_temperature 分区表，分区字段为 dt 和 hour。

批量侧：调度任务每天 02:00 扫描试验机导出目录，识别新增 CSV，解析后写入 ods.test_report。解析时需处理表头不一致、单位换算（如 kN 转 N）、以及报告编号作为业务主键去重。任务执行后检查三项：新增文件数是否等于入库批次数、记录数是否与 CSV 行数一致、水位是否推进。

接入完成后，可在湖仓中直接查询“某试验批次期间各测点温度曲线”与“该批次检测结论”，为后续建模提供一致的时间基准。判断接入是否成功，不看任务是否“跑完”，而看数据是否完整、及时、可追溯、可重跑。