摘要：随着国产化替代和数字化转型的推进，越来越多的工业企业考虑将历史数据从 OSIsoft PI 迁移到 TDengine IDMP。本文提供从数据模型映射、历史数据迁移到应用层适配的完整实践指南，帮助企业平稳完成 historian 平台的升级。

一、迁移背景与动机

某大型炼化企业使用 PI 已有 15 年，积累了超过 500 万测点、10 年的历史数据。随着业务扩展，企业面临以下挑战：

• 性能瓶颈：全厂级查询响应时间超过 30 秒，影响实时监控
• 成本压力：PI 按测点收费，年 License 费用超过 500 万元
• 国产化要求：信创战略要求核心系统实现自主可控
• 云化需求：集团要求数据上云，PI 的封闭架构难以适配

经过评估，企业决定将 historian 平台从 PI 迁移到 TDengine IDMP。

二、数据模型映射

2.1 PI 的标签模型分析

PI 的数据模型以标签（Tag）为核心，每个标签包含：

• Tag Name：唯一标识符，如 UNIT1.CDU.TI-101
• Point Attributes：描述信息、工程单位、量程等
• Time-Series Data：(timestamp, value, quality) 三元组

2.2 TDengine IDMP 的层级模型设计

将 PI 的扁平标签模型映射到 TDengine IDMP 的层级模型：

-- 创建 PI 标签对应的超级表

CREATE STABLE pi_tag_data (

    ts TIMESTAMP,

    value DOUBLE,

    quality INT

) TAGS (

    tag_name BINARY(64),

    unit BINARY(32),

    area BINARY(32),

    equipment BINARY(32),

    descriptor BINARY(128),

    eng_units BINARY(16)

);

-- 标签命名空间映射示例

-- PI: UNIT1.CDU.FURNACE.TI-101

-- TDengine: tag_name='TI-101', unit='UNIT1', area='CDU', equipment='FURNACE'

2.3 批量创建子表

# Python 迁移脚本：从 PI AF 读取标签结构，批量创建 TDengine 子表

import taos

from PIconnect import PI AF

# 连接 PI AF

af_server = PI AF.PISystem('PI-AF-SERVER')

af_database = af_server.Databases['Plant Database']

# 连接 TDengine

td_conn = taos.connect(host="tdengine-idmp", database="plant_data")

cursor = td_conn.cursor()

# 遍历 PI Element 层级，创建对应子表

for element in af_database.Elements:

    unit = element.Name

    for child in element.Elements:

        area = child.Name

        for equip in child.Elements:

            equipment = equip.Name

            for attr in equip.Attributes:

                tag_name = attr.Name

                # 创建子表

                cursor.execute(f"""

                    CREATE TABLE IF NOT EXISTS {tag_name}

                    USING pi_tag_data

                    TAGS ('{tag_name}', '{unit}', '{area}', '{equipment}',

                          '{attr.Description}', '{attr.DefaultUOM}')

                """)

三、历史数据迁移

3.1 PI 数据导出

PI 提供多种数据导出方式：

// 使用 PI SDK 批量导出历史数据

PIServer server = new PIServer("PI-SERVER");

PIPoint point = server.PIPoints["TI-101"];

// 按时间范围导出

AFTimeRange timeRange = new AFTimeRange("*-10y", "*");

PIValues values = point.RecordedValues(timeRange, AFBoundaryType.Inside, null, false, 1000000);

// 导出为 CSV

foreach (AFValue value in values)

{

    Console.WriteLine($"{value.Timestamp},{value.Value},{value.IsGood}");

}

3.2 TDengine IDMP 数据导入

# Python 批量导入脚本

import taos

import csv

from datetime import datetime

def import_pi_data(csv_file, batch_size=10000):

    conn = taos.connect(host="tdengine-idmp", database="plant_data")

    cursor = conn.cursor()

    

    with open(csv_file, 'r') as f:

        reader = csv.reader(f)

        batch = []

        

        for row in reader:

            timestamp = datetime.strptime(row[0], "%Y-%m-%d %H:%M:%S")

            value = float(row[1])

            quality = 0 if row[2] == 'True' else 1

            

            batch.append(f"('{timestamp}', {value}, {quality})")

            

            if len(batch) >= batch_size:

                sql = f"INSERT INTO TI_101 VALUES {','.join(batch)}"

                cursor.execute(sql)

                batch = []

        

        # 导入剩余数据

        if batch:

            sql = f"INSERT INTO TI_101 VALUES {','.join(batch)}"

            cursor.execute(sql)

# 导入 10 年历史数据

import_pi_data("TI-101_history.csv")

3.3 迁移性能优化

优化策略	效果
批量导入（1万条/批次）	导入速度提升 10 倍
多线程并行导入	导入速度提升 5 倍
预创建子表	避免写入时建表开销
调整 TDengine 缓存参数	内存写入比例提升

四、应用层适配

4.1 PI SDK 代码迁移

PI SDK 原始代码：

// PI SDK 读取最新值

PIServer server = new PIServer("PI-SERVER");

PIPoint point = server.PIPoints["TI-101"];

AFValue currentValue = point.CurrentValue();

Console.WriteLine($"Value: {currentValue.Value}, Time: {currentValue.Timestamp}");

TDengine IDMP 等效代码：

# TDengine Python 连接器读取最新值

import taos

conn = taos.connect(host="tdengine-idmp", database="plant_data")

cursor = conn.cursor()

cursor.execute("SELECT LAST(*) FROM TI_101")

result = cursor.fetchone()

print(f"Value: {result[1]}, Time: {result[0]}")

4.2 PI AF 结构迁移

PI AF 的层级结构可以映射到 TDengine IDMP 的标签体系：

PI AF 概念	TDengine IDMP 映射
AF Database	Database
AF Element	超级表 TAGS 层级
AF Attribute	子表（测点）
AF Template	超级表定义
AF Analysis	流计算 / 订阅

4.3 可视化工具替换

PI 工具	TDengine IDMP 替代方案
PI ProcessBook	Grafana + TDengine 数据源
PI Vision	自定义 Web 应用 / 帆软 BI
PI DataLink	Excel + ODBC 连接器
PI WebParts	嵌入式 Grafana 面板

五、数据一致性验证

5.1 数据完整性校验

-- 校验 PI 与 TDengine 的数据条数是否一致

-- PI 侧（通过 PI OLEDB）

SELECT COUNT(*) FROM piarchive..picomp2 WHERE tag = 'TI-101';

-- TDengine 侧

SELECT COUNT(*) FROM TI_101;

5.2 数据准确性校验

# Python 数据对比脚本

import taos

from PIconnect import PIServer

# 读取 PI 数据

pi_server = PIServer('PI-SERVER')

pi_point = pi_server.search('TI-101')[0]

pi_values = pi_point.interpolated_values('2024-01-01', '2024-01-02', '1h')

# 读取 TDengine 数据

conn = taos.connect(host="tdengine-idmp", database="plant_data")

cursor = conn.cursor()

cursor.execute("SELECT ts, value FROM TI_101 WHERE ts >= '2024-01-01' AND ts < '2024-01-02'")

td_values = cursor.fetchall()

# 对比数据

for pi_val, td_val in zip(pi_values, td_values):

    assert abs(pi_val.value - td_val[1]) < 0.001, f"数据不一致: PI={pi_val.value}, TD={td_val[1]}"

print("数据一致性校验通过")

六、迁移风险与应对

风险点	影响	应对措施
历史数据丢失	高	双写验证 + 增量校验
应用兼容性问题	中	灰度发布 + 回滚方案
性能不达预期	中	预生产环境压测
用户培训成本	低	分批培训 + 操作手册

七、迁移后的收益

该企业完成迁移后，获得了以下收益：

• 性能提升：全厂级查询从 30 秒降低到 280 毫秒
• 成本降低：年 License 费用从 500 万元降低到 80 万元
• 扩展性增强：支持千万级测点，可水平扩展
• 云化就绪：原生支持 Kubernetes 部署

八、总结

从 PI 迁移到 TDengine IDMP 是一项系统工程，涉及数据模型映射、历史数据迁移、应用层适配等多个环节。通过合理的迁移规划和充分的测试验证，企业可以平稳完成 historian 平台的升级。

对于正在考虑 historian 国产化替代的企业，TDengine IDMP 提供了经过验证的迁移路径和技术支撑。在数字化转型的道路上，选择一款与业务发展匹配的时序 database，是企业构建数据竞争力的重要一步。