构建高并发AI数据流水线：基于腾讯云CKafka+DataPipeline+TI-ONE的实战

┌─────────────┐     ┌──────────────────┐     ┌────────────────┐
│  业务DB     │────▶│  CDC (Debezium)  │────▶│  CKafka 集群   │
│  (TDSQL)    │     │   + 轻量清洗     │     │  (30分区, 3副本)│
└─────────────┘     └──────────────────┘     └───────┬────────┘
                                                       │
┌─────────────┐     ┌──────────────────┐              │
│  埋点日志   │────▶│  CLS 日志投递   │──────────────┘
│  (Nginx)    │     │   (自定义解析)   │
└─────────────┘     └──────────────────┘
                                                       ▼
                                            ┌─────────────────────┐
                                            │  DataPipeline 实时  │
                                            │  - 动态Schema注册   │
                                            │  - 特征计算(UDTF)  │
                                            │  - 窗口聚合(滑动)  │
                                            └──────────┬──────────┘
                                                       │
                                                       ▼
                                            ┌─────────────────────┐
                                            │  特征存储(Redis)   │
                                            │  + 样本湖(CFS)     │
                                            └──────────┬──────────┘
                                                       │
                                  ┌────────────────────┼────────────────────┐
                                  ▼                    ▼                    ▼
                          ┌──────────────┐  ┌──────────────┐  ┌──────────────┐
                          │ TI-ONE 训练  │  │ 在线推理服务 │  │ 离线回放任务 │
                          │ (PyTorch)    │  │ (Triton)    │  │ (Flink)      │
                          └──────────────┘  └──────────────┘  └──────────────┘

tccli ckafka CreateTopic \
  --InstanceId ckafka-xxxxx \
  --TopicName ai_feature_raw \
  --PartitionNum 30 \
  --ReplicaNum 3 \
  --RetentionMs 604800000 \
  --SegmentMs 3600000 \
  --MaxMessageBytes 10485760 \
  --CleanUpPolicy delete

from confluent_kafka import Producer, KafkaError
import json
import time
from threading import Thread
from queue import Queue

class AsyncBatchProducer:
    def __init__(self, brokers, topic, batch_size=500, linger_ms=100):
        self.conf = {
            'bootstrap.servers': brokers,
            'compression.type': 'lz4',
            'batch.num.messages': batch_size,
            'linger.ms': linger_ms,
            'queue.buffering.max.messages': 50000,
            'enable.idempotence': True,
            'max.in.flight.requests.per.connection': 5,  # 允许乱序但幂等保证
            'acks': 'all',
            'retries': 3,
        }
        self.producer = Producer(self.conf)
        self.topic = topic
        self.queue = Queue(maxsize=10000)
        self.running = True
        # 启动后台发送线程
        Thread(target=self._sender, daemon=True).start()

    def _sender(self):
        while self.running:
            msgs = []
            # 批量取出
            while len(msgs) < 500 and not self.queue.empty():
                msgs.append(self.queue.get())
            if msgs:
                # 按分区 Key 聚合（保证同一用户顺序）
                for key, value in msgs:
                    self.producer.produce(
                        self.topic,
                        key=key.encode('utf-8'),
                        value=json.dumps(value, ensure_ascii=False).encode('utf-8'),
                        callback=self._delivery_report
                    )
                self.producer.flush(timeout=2)

    def _delivery_report(self, err, msg):
        if err:
            print(f"Delivery failed: {err}")
        # 可记录消息偏移量到外部存储

    def send(self, key, data):
        self.queue.put((key, data))

    def close(self):
        self.running = False
        self.producer.flush()

# 自定义 Python Transform (DataPipeline 官方支持)
from typing import Dict, Any
import jsonschema
from schemaregistry import SchemaRegistryClient  # 自建或使用腾讯云 TCR

client = SchemaRegistryClient(endpoint="http://schema-svc:8081")

def transform(record: Dict[str, Any]) -> Dict[str, Any]:
    # 从消息头获取 schema_id
    schema_id = record.get("_schema_id", "default_v1")
    schema = client.get_schema(schema_id)
    
    # 校验并抽取特征
    try:
        jsonschema.validate(record["payload"], schema)
    except jsonschema.ValidationError as e:
        # 若校验失败，尝试自动演化为新版本
        new_schema = auto_evolve(schema, record["payload"])
        client.register_schema(f"{schema_id}_v{version+1}", new_schema)
        # 重新校验
        jsonschema.validate(record["payload"], new_schema)
    
    # 提取特征字段（只保留模型需要的 80 个特征，过滤冗余）
    features = {k: v for k, v in record["payload"].items() if k in FEATURE_WHITELIST}
    # 添加衍生特征（例如 hour_of_day）
    features["hour"] = time.localtime(features["event_ts"]).tm_hour
    return {"features": features, "label": record["payload"].get("click", 0)}

-- DataPipeline 支持类 SQL 的窗口定义
CREATE STREAM user_clicks (
    user_id STRING,
    item_id STRING,
    event_ts TIMESTAMP,
    action STRING
) WITH (KAFKA_TOPIC='ai_feature_raw', VALUE_FORMAT='JSON');

-- 滑动窗口：每 10 秒计算一次过去 30 秒的 CTR
CREATE TABLE user_ctr_30s AS
SELECT
    user_id,
    COUNT_IF(action='click') / COUNT(*) AS ctr,
    TUMBLE_END(event_ts, INTERVAL '10' SECOND) AS window_end
FROM user_clicks
GROUP BY
    user_id,
    TUMBLE(event_ts, INTERVAL '10' SECOND);

from redis import Redis
redis_client = Redis(host='redis-cluster', decode_responses=True)

def process_batch(messages):
    # 1. 处理消息，更新 Redis 状态（使用 Lua 脚本保证原子性）
    lua_script = """
        local ctr_key = KEYS[1]
        local count = redis.call('INCR', ctr_key)
        if count > 100 then
            redis.call('EXPIRE', ctr_key, 60)
        end
        return count
    """
    for msg in messages:
        redis_client.eval(lua_script, 1, f"ctr:{msg['user_id']}", 1)
    # 2. 手动提交 Kafka Offset（原子性依赖幂等）
    consumer.commit()

/samples/
  dt=2026-08-22/
    hour=00/ part-0001.parquet
    hour=01/ part-0002.parquet
  dt=2026-08-21/ ...

from tencentcloud.tione.v20211111 import tione_client, models
import json

# 初始化客户端
cred = credential.Credential(os.environ['SECRET_ID'], os.environ['SECRET_KEY'])
client = tione_client.TioneClient(cred, "ap-guangzhou")

# 构造训练任务
req = models.CreateTrainingTaskRequest()
req.Name = "ctr_model_v2.3"
req.FrameworkName = "PYTORCH"
req.FrameworkVersion = "1.13"
req.TrainingMode = "DDP"  # 分布式
req.ResourceConfig = {
    "InstanceType": "TI.GN10X.2XLARGE40",  # A100 单卡
    "InstanceCount": 4,
    "Cpu": 16,
    "Memory": 64
}
req.CodePackagePath = "cos://bucket/code/train.py"
req.DataConfigs = [{
    "DataSourceType": "CFS",
    "MappingPath": "/data/samples",
    "CFSConfig": {
        "Id": "cfs-xxxxx",
        "Path": "/samples"
    }
}]
req.Outputs = [{
    "OutputPath": "cos://bucket/models/ctr_v2.3"
}]
req.HyperParameters = json.dumps({
    "batch_size": 2048,
    "lr": 0.001,
    "dropout": 0.2,
    "feature_version": "v2.3"
})

resp = client.CreateTrainingTask(req)
print(f"Task ID: {resp.TaskId}")

from confluent_kafka import Consumer, TopicPartition

def seek_to_timestamp(consumer, topic, timestamp_ms):
    partitions = consumer.list_topics(topic).topics[topic].partitions
    tps = []
    for p in partitions:
        tp = TopicPartition(topic, p)
        # 获取该分区指定时间戳的偏移量
        offset = consumer.offsets_for_times([TopicPartition(topic, p, timestamp_ms)])[0]
        tps.append(TopicPartition(topic, p, offset.offset))
    consumer.assign(tps)
    return consumer

# 回放 2026-08-21 00:00:00 到 2026-08-22 00:00:00
start_ts = int(datetime(2026,8,21).timestamp() * 1000)
end_ts = int(datetime(2026,8,22).timestamp() * 1000)
consumer = Consumer({'bootstrap.servers': 'ckafka-xxx', 'group.id': 'replay_group'})
seek_to_timestamp(consumer, 'ai_feature_raw', start_ts)
# 然后循环消费，直到消息时间戳 > end_ts
while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        break
    ts = msg.timestamp()[1]  # 业务时间戳（需在生产者设置）
    if ts > end_ts:
        break
    process_sample(msg.value())

name: "ctr_ensemble"
platform: "ensemble"
input [
  { name: "user_id", data_type: TYPE_STRING, dims: [1] },
  { name: "item_id", data_type: TYPE_STRING, dims: [1] },
  { name: "context", data_type: TYPE_FP32, dims: [10] }
]
output [
  { name: "score", data_type: TYPE_FP32, dims: [1] }
]
ensemble_scheduling {
  step [
    { model_name: "feature_transform", model_version: -1, input_map { key: "uid" value: "user_id" } },
    { model_name: "embedding_lookup", model_version: -1, input_map { key: "item" value: "item_id" } },
    { model_name: "ctr_model", model_version: -1, input_map { key: "emb" value: "embedding_lookup/emb_out" } }
  ]
}

# 训练侧保存摘要
import hashlib
def feature_hash(features):
    sorted_items = sorted(features.items())
    s = "|".join([f"{k}={v}" for k,v in sorted_items])
    return hashlib.md5(s.encode()).hexdigest()
# 推理侧同样计算，并写入 CLS

consumer_conf['partition.assignment.strategy'] = 'cooperative-sticky'

if redis_client.info()['used_memory_peak'] > 0.9 * maxmemory:
    time.sleep(0.1)  # 降低消费速度