云原生实践：在华为云上基于 TDengine 构建 AI 数据平台

引言

云计算已成为企业数字化转型的标准基础设施。在云原生架构下，应用被设计为可在云环境中弹性扩展、快速部署和高效运维。对于 AI 应用而言，云原生架构提供了弹性计算资源、分布式存储能力和自动化运维工具，极大地加速了 AI 项目的落地。

TDengine 作为国产时序 database，全面拥抱云原生架构，支持容器化部署、Kubernetes 编排、存储计算分离和多云互联。在华为云生态中，TDengine 可以与 IoTDA、ModelArts、CCE 等服务深度集成，构建完整的云原生 AI 数据平台。

本文将从云原生架构设计、华为云集成实践、AI 数据 pipeline 构建等多个维度，展示如何在华为云上基于 TDengine 构建企业级 AI 数据平台。

一、云原生 AI 数据平台的架构设计

1.1 整体架构

一个典型的云原生 AI 数据平台包括以下层次：

· 数据采集层：物联网设备、传感器、日志系统等数据源

· 数据接入层：华为云 IoTDA、DIS 等数据接入服务

· 数据存储层：TDengine 时序数据库集群

· 特征工程层：TDengine 流计算引擎

· 模型训练层：华为云 ModelArts、自建 GPU 集群

· 模型推理层：华为云 FunctionGraph、容器化推理服务

· 应用展示层：Web 应用、移动端、大屏展示

1.2 核心设计原则

弹性伸缩

根据业务负载自动调整资源规模。在数据高峰期自动扩容，在低谷期自动缩容，实现资源利用率最大化。

高可用性

通过多副本、故障自动切换等机制，确保系统 7×24 小时稳定运行。单点故障不影响整体服务。

数据一致性

在分布式环境下保证数据的一致性和可靠性。TDengine 采用 Raft 一致性协议，确保数据多副本同步。

可观测性

通过日志、指标、追踪等手段，全面监控系统的运行状态，快速发现和定位问题。

二、华为云上的 TDengine 部署实践

2.1 通过云市场镜像快速部署

华为云云市场提供了 TDengine 的官方镜像，用户可以通过镜像快速创建 TDengine 实例。

# 登录华为云控制台

# 进入云市场，搜索 "TDengine"

# 选择合适的镜像版本，创建云服务器实例

# 实例创建完成后，TDengine 服务自动启动

2.2 通过 Kubernetes Operator 部署集群

对于需要高可用和弹性伸缩的场景，建议使用 Kubernetes 部署 TDengine 集群。

# tdengine-cluster.yaml

apiVersion: "tdengine.com/v1"

kind: "TDengineCluster"

metadata:

  name: "ai-data-cluster"

spec:

  replicas: 3

  resources:

    requests:

      memory: "4Gi"

      cpu: "2"

    limits:

      memory: "8Gi"

      cpu: "4"

  storage:

    size: "100Gi"

    storageClassName: "csi-disk"

# 部署 TDengine 集群

kubectl apply -f tdengine-cluster.yaml

# 查看集群状态

kubectl get tdenginecluster

2.3 存储计算分离架构

在华为云上，可以利用云硬盘（EVS）和对象存储（OBS）实现存储计算分离：

· 热数据：存储在 EVS 高性能云硬盘上，保证查询速度

· 温数据：存储在 EVS 标准云硬盘上，平衡性能和成本

· 冷数据：自动转存至 OBS 对象存储，实现近乎无限的存储扩展

-- 创建数据库时配置分层存储

CREATE DATABASE ai_data

  DURATION 7d

  KEEP 365d

  COMP 2;

三、与华为云 IoTDA 的集成

3.1 设备数据接入

华为云 IoTDA 提供了设备接入和管理能力。设备数据通过 IoTDA 接入后，可以通过规则引擎实时流转至 TDengine。

// IoTDA 规则引擎配置

{

  "rule_name": "sensor_data_to_tdengine",

  "sql": "SELECT * FROM device_message WHERE device_type = 'sensor'",

  "actions": [

    {

      "type": "DIS",

      "stream_name": "sensor_data_stream"

    }

  ]

}

3.2 数据流转至 TDengine

通过华为云 DIS（数据接入服务），将设备数据实时流转至 TDengine：

import taos

from huaweicloudsdkdis import DisClient

# 连接 TDengine

conn = taos.connect(host="tdengine-cluster", database="iot_data")

# 消费 DIS 数据流

dis_client = DisClient(region="cn-north-4")

for record in dis_client.get_records("sensor_data_stream"):

    device_id = record['device_id']

    temperature = record['temperature']

    vibration = record['vibration']

    # 写入 TDengine

    cursor = conn.cursor()

    cursor.execute(f"""

        INSERT INTO d_{device_id} USING sensor_data TAGS ('{device_id}', 'type_a')

        VALUES (NOW, {temperature}, {vibration})

    """)

四、AI 数据 Pipeline 构建

4.1 实时特征工程

使用 TDengine 的流计算引擎，实时生成 AI 特征：

-- 创建实时特征流

CREATE STREAM device_features

INTO TABLE feature_table

AS

SELECT

    _irowts AS ts,

    device_id,

    AVG(temperature) AS temp_mean,

    STDDEV(temperature) AS temp_std,

    MAX(vibration) AS vib_max,

    ROOT_MEAN_SQUARE(vibration) AS vib_rms

FROM sensor_data

PARTITION BY device_id

INTERVAL(5m);

4.2 模型训练与部署

通过华为云 ModelArts，使用 TDengine 中的数据进行模型训练：

from modelarts.session import Session

from modelarts.estimator import Estimator

# 创建 ModelArts 会话

session = Session()

# 定义训练作业

estimator = Estimator(

    modelarts_session=session,

    framework_type='PyTorch',

    framework_version='1.9',

    code_dir='s3://ai-bucket/training-code/',

    boot_file='train.py',

    hyperparameters={

        'data_source': 'tdengine://tdengine-cluster/ai_data/feature_table',

        'epochs': 100,

        'batch_size': 32

    },

    output_path='s3://ai-bucket/model-output/',

    train_instance_type='modelarts.p3.large',

    train_instance_count=1

)

# 启动训练

estimator.fit()

4.3 模型推理服务

将训练好的模型部署为华为云 FunctionGraph 函数，实现实时推理：

import json

import taos

import joblib

import numpy as np

# 加载模型

model = joblib.load('/opt/model/fault_detection.pkl')

# 连接 TDengine

conn = taos.connect(host="tdengine-cluster", database="ai_data")

def handler(event, context):

    """FunctionGraph 处理函数"""

    device_id = event['device_id']

    # 从 TDengine 读取最新特征

    cursor = conn.cursor()

    cursor.execute(f"""

        SELECT temp_mean, temp_std, vib_max, vib_rms

        FROM feature_table

        WHERE device_id = '{device_id}'

        ORDER BY ts DESC LIMIT 1

    """)

    row = cursor.fetchone()

    # 模型推理

    features = np.array([row])

    prediction = model.predict(features)[0]

    probability = model.predict_proba(features)[0][1]

    return {

        'device_id': device_id,

        'prediction': int(prediction),

        'probability': float(probability)

    }

五、监控与运维

5.1 集群监控

通过华为云 AOM（应用运维管理）监控 TDengine 集群的运行状态：

# Prometheus 监控配置

apiVersion: v1

kind: ConfigMap

metadata:

  name: tdengine-prometheus-config

data:

  prometheus.yml: |

    global:

      scrape_interval: 15s

    scrape_configs:

      - job_name: 'tdengine'

        static_configs:

          - targets: ['tdengine-cluster:6043']

5.2 告警配置

配置告警规则，当关键指标异常时及时通知运维人员：

# 告警规则

- alert: TDengineHighMemoryUsage

  expr: tdengine_memory_usage / tdengine_memory_total > 0.8

  for: 5m

  labels:

    severity: warning

  annotations:

    summary: "TDengine 内存使用率过高"

    description: "实例 {{ $labels.instance }} 内存使用率超过 80%"

六、最佳实践

6.1 数据分区策略

合理的数据分区策略可以提升查询性能和数据管理效率：

-- 按时间自动分区

CREATE DATABASE ai_data

  DURATION 7d    -- 每个分区保留 7 天数据

  KEEP 365d      -- 总保留 365 天数据

  COMP 2;        -- 压缩级别

6.2 读写分离

对于读多写少的场景，可以采用读写分离架构：

· 写入节点：负责接收数据写入

· 查询节点：负责处理查询请求

· 副本节点：提供数据冗余和故障恢复

6.3 备份与恢复

定期进行数据备份，确保数据安全：

# 使用 taosdump 进行数据备份

taosdump -o /backup/ai_data_$(date +%Y%m%d) ai_data

# 数据恢复

taosdump -i /backup/ai_data_20240101 ai_data

七、结语

在华为云生态中，TDengine 可以与 IoTDA、ModelArts、CCE、DIS 等服务深度集成，构建完整的云原生 AI 数据平台。从设备数据接入到实时特征工程，从模型训练到在线推理，TDengine 在每个环节都提供了有力的支持。通过合理的架构设计和最佳实践，企业可以在华为云上快速构建高可用、高性能、可扩展的 AI 数据基础设施，加速 AI 应用的落地和价值实现。