3、数据源接入：ERP/SCM/WMS/TMS数据接口规范、实时流与批处理数据融合

供应链控制塔的核心价值在于打破系统孤岛，将分散在 ERP、SCM、WMS、TMS 中的业务数据汇聚到统一的可视化层。本章聚焦数据源接入的技术规范与融合策略，涵盖接口协议、数据模型适配、实时流与批处理数据的协同处理，以及常见集成避坑指南。

3.1 四大核心系统数据接口规范

控制塔需要对接的企业内部系统通常包括企业资源计划（ERP）、供应链管理（SCM）、仓库管理系统（WMS）和运输管理系统（TMS）。各系统的数据特征与接口规范差异显著，下表总结了关键差异点：

3.2 实时流与批处理数据融合架构

控制塔需要同时处理高频率的实时事件（如扫码出库、GPS 位置更新）和低频但数据量大的批处理数据（如每日财务结算、月度需求计划）。下图展示了典型的 Lambda 架构在控制塔数据融合中的应用：

3.3 数据融合的关键步骤与代码示例

实时流与批处理数据的融合通常遵循以下步骤：

1. 数据采集：通过 CDC（Change Data Capture）工具或消息队列监听实时变更，同时通过定时任务拉取批处理文件。
2. 数据标准化：将不同系统的字段映射为统一的数据模型（如统一订单状态码、时间格式、单位）。
3. 数据清洗与去重：处理重复数据、缺失值、异常值，确保数据质量。
4. 数据关联与合并：以业务主键（如订单号、运单号）为纽带，将实时流数据与批处理历史数据关联。
5. 数据写入目标存储：将融合后的数据写入 ClickHouse、Elasticsearch 或 Druid 等支持实时查询的存储引擎。

以下是一个使用 Python 和 Apache Kafka 进行实时流与批处理数据融合的简化示例：

# 示例：实时流与批处理数据融合（伪代码）
from kafka import KafkaConsumer
import pandas as pd
from datetime import datetime

# 1. 初始化 Kafka 消费者（实时流）
consumer = KafkaConsumer(
    'wms_inventory_events',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='latest',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

# 2. 加载批处理数据（每日库存快照）
batch_df = pd.read_parquet('hdfs://data/warehouse/inventory_snapshot/dt=2025-03-01/')

# 3. 实时流处理与融合
for message in consumer:
    event = message.value
    # 实时事件：库存扣减
    sku = event['sku']
    qty_change = event['quantity_change']
    warehouse = event['warehouse_id']
    
    # 从批处理数据中获取该 SKU 的初始库存
    initial_qty = batch_df.loc[
        (batch_df['sku'] == sku) & (batch_df['warehouse_id'] == warehouse),
        'quantity'
    ].iloc[0]
    
    # 融合计算：当前实时库存 = 批处理初始库存 + 实时变更累计
    current_qty = initial_qty + qty_change
    
    # 写入实时查询引擎（如 ClickHouse）
    write_to_clickhouse(sku, warehouse, current_qty, datetime.now())

3.4 接口适配与兼容性清单

在实际项目中，控制塔需要适配多种接口协议和数据格式。下表列出了常见适配场景及推荐方案：

3.5 数据质量与监控

数据源接入后，必须建立数据质量监控机制，确保控制塔展示的数据准确可信。推荐监控以下指标：

• 数据延迟：实时流从产生到进入控制塔的端到端延迟（目标：< 5 秒）。
• 数据完整性：批处理数据是否完整到达，是否有缺失分区或文件。
• 数据一致性：实时流与批处理数据在关键指标（如库存总量）上是否一致。
• 接口可用性：各系统接口的健康状态与响应时间。