4. 集成模式与数据流设计

在 PI System 与数据湖分析平台集成时，真正决定系统可用性和数据价值的，往往不是“能不能把数据搬过去”，而是“用哪种模式搬、按什么节奏搬、在什么条件下触发、端到端链路如何保证一致”。PI System 侧的数据源包括 PI Data Archive 中的时序测点、PI Asset Framework（AF）中的资产模型与层级、PI Event Frames 中的事件记录，以及 PI Interfaces/PI Connectors 采集的实时值。数据湖侧则可能是对象存储（如 S3/OSS/HDFS）、湖表格式（Iceberg/Hudi/Delta）、消息队列（Kafka）或流计算引擎（Flink/Spark Structured Streaming）。不同模式的选择，本质上是在时效性、吞吐、成本、复杂度和一致性之间做权衡。

四种典型集成模式及其适用边界

批处理（Batch）：按固定周期（如每小时、每天）将一段时间窗口内的 PI 数据抽取、转换并写入数据湖。典型实现是 PI Web API 的 streams/{webId}/recorded 或 interpolated 接口，或 PI Integrator for Business Analytics 的定时发布任务。它适合历史回填、T+1 报表、离线特征工程、大规模重算。优点是吞吐高、对源系统压力可控、易于重跑；缺点是时效性差，无法支撑秒级告警或实时看板。

流式（Streaming）：数据以持续不断的方式从 PI 侧推送到消息队列或流处理引擎，端到端延迟通常在秒级甚至亚秒级。常见路径是 PI Event Frames 或 PI AF 分析触发后，通过 PI Web API 的 Channels/WebSocket 或 PI Adapter 推送到 Kafka，再由 Flink 消费写入湖表。它适合实时监控、异常检测、在线特征计算。代价是运维复杂、乱序与重复处理需要额外机制、对源系统连接稳定性要求高。

微批（Micro-batch）：介于批与流之间，以较小时间窗口（如 1 分钟、5 分钟）周期性触发，典型代表是 Spark Structured Streaming 的 trigger 或 Flink 的 checkpoint 间隔。它兼顾了较低延迟与较高吞吐，适合准实时指标聚合、分钟级 KPI 入湖。选择微批时，窗口大小要与数据湖小文件治理策略协同，否则容易产生大量小文件。

事件驱动（Event-driven）：不以时间为触发条件，而以业务事件为触发条件，例如 PI Event Frames 生成、AF 分析结果变化、设备状态跳变、批次开始/结束。事件通过消息总线分发，下游按需消费。它适合“有事件才处理”的场景，能显著降低无效计算和存储。事件驱动常与流式结合，但触发语义不同：流式强调持续，事件驱动强调条件。

端到端数据流设计方法

端到端设计可以按“源—抽取—传输—缓冲—处理—落湖—服务”七段来梳理，但每一段都要落到具体对象和参数上。以某电厂 PI System 到数据湖的集成为例：源端是 PI Data Archive 中的 Plant1.Boiler1.Temp 等测点，以及 AF 中 Boiler1 资产下的属性；抽取端使用 PI Web API 按时间范围拉取，或使用 PI Adapter 推送；传输端进入 Kafka 主题 pi.raw.timeseries；缓冲端保留 7 天；处理端用 Flink 做去重、单位换算、质量码过滤；落湖端写入 Iceberg 表 ods_pi_timeseries，按 asset_id 和 event_date 分区；服务端供 BI 与特征平台查询。

设计时最容易被忽略的是时间语义与水位线。PI 数据带有 Source Timestamp 和 Server Timestamp，前者是现场采集时间，后者是 PI 服务器接收时间。入湖时应明确以哪个时间做事件时间，并设置水位线容忍乱序。例如允许 5 分钟乱序，则水位线为 max_event_time - 5min，超过水位线的迟到数据进入侧输出流单独处理。若不做这层设计，流式聚合结果会因迟到数据反复修正，导致下游指标不稳定。

第二个关键是幂等与去重。PI 数据在重试、断点续传或回填时可能重复。常用做法是以 webId + timestamp + quality 作为业务主键，在湖表上做 merge/upsert，或在 Flink 中用 keyed state 做去重。批处理场景则通过分区覆盖（overwrite partition）保证重跑一致性。

第三个关键是质量码处理。PI 的 quality 字段包含 Good、Bad、Questionable 等状态。入湖时不应直接丢弃 Bad 数据，而应保留并打标，供下游按需过滤。例如在 Flink 中把 quality != "Good" 的记录写入 ods_pi_timeseries_bad 表，正常数据写入主表，这样既保证主表干净，又不丢失可追溯性。

组合模式与端到端一致性

真实项目很少只用一种模式。一个可落地的组合是：每日凌晨用批处理回填前一天的全量历史数据，保证完整性；白天用微批每 5 分钟增量入湖，保证准实时；对关键设备跳变和 Event Frame 用事件驱动实时推送，保证告警时效。三条链路写入同一张 Iceberg 表时，必须统一主键和分区策略，否则会出现重复或覆盖冲突。建议在湖表上建立 merge_key = webId + timestamp，并让批、微批、事件三条链路都走同一套 upsert 逻辑。

端到端一致性还需要关注断点续传。PI Web API 拉取时记录最后成功的时间戳，下次从该时间戳继续；Kafka 消费端提交 offset 与湖表写入应在同一事务或幂等框架下完成。若使用 Flink，可开启 checkpoint 并将 offset 与状态一起持久化，恢复时从最近 checkpoint 继续，避免数据丢失或重复。

最后，模式选择不是一次性决策。随着业务从“看历史”转向“看实时”，再从“看实时”转向“按事件响应”，集成模式也应随之演进。建议在架构上保留统一的消息层和湖表层，让批、微批、流式、事件驱动四种模式可以按需切换或并存，而不是把某一种模式写死在采集脚本里。这样，当业务提出“把延迟从 5 分钟降到 30 秒”时，你只需要调整触发器和窗口参数，而不必重建整条链路。