智能驾驶全闭环数据平台解决方案

一、方案背景

智能驾驶行业的数据挑战

自动驾驶技术的核心竞争力在于数据——谁能更快地将海量行驶数据转化为高质量训练数据并迭代模型，谁就能在竞争中领先。然而，现实中智能驾驶企业面临的数据挑战极为复杂：

数据来源多样、规模巨大

• 单辆路测车每小时产生数 GB 数据（摄像头标注 + LiDAR 点云 + CAN 总线）
• 量产车队规模达百万辆，高频遥测 QPS 峰值 10 万~100 万 msg/s（参考吉利 AutoMQ 案例）
• 数据格式异构：结构化时序（CAN 信号）、半结构化（JSON 事件）、大文件（Parquet 标注）

数据闭环路径长、链路断裂 从路测采集到模型更新上车，中间需要经历标注、仿真扩增、训练集构建、离线评测、OTA 灰度、遥测回流等多个环节，每个环节往往由不同团队使用不同工具完成，数据在孤岛之间传递，效率低下。

长尾场景难以覆盖 极端天气、稀有障碍物、特殊交通场景（NEAR_MISS、接管）在真实采集中出现概率极低，但对安全性验证至关重要。靠人工收集长尾数据成本极高、周期极长。

合规与安全要求严格 监管机构对自动驾驶的安全评估、OTA 升级、故障追溯有明确要求，需要完整的数据证据链和风险事件记录。

二、传统方案的痛点

架构碎片化，工具链臃肿

以一个典型的智驾数据团队为例，维护上述工具链需要 Spark 集群 + Flink 集群 + 消息队列 + 多个数据库 + LLM 服务，基础设施成本和运维负担极重，数据工程师的大量精力耗费在管道维护而非业务价值上。

数据闭环周期长

传统架构下，从量产车上传一个接管事件到该事件进入下一轮训练集，往往需要 2~4 周：

接管事件上报 ↓（T+1 批量处理） 数仓入库 ↓（人工排期） 标注任务分配 ↓（3~5 天） 标注完成审核 ↓（人工构建） 训练集版本打包 ↓（提交集群） 模型训练评估 ↓（部署审批） OTA 上线

国内头部车企新势力已将数据闭环周期压缩到数周以内，传统架构已成为瓶颈。

长尾场景补充能力弱

仿真系统与数据平台相互独立，真实采集的边缘场景无法自动注入仿真场景库，导致长尾场景覆盖率低，模型在极端情况下表现差。

三、ClickZetta Lakehouse 方案

方案架构

本方案基于 ClickZetta Lakehouse 构建一体化智能驾驶数据平台，覆盖研发、测试、量产运营三个阶段，通过 Data Flywheel 将量产车行驶数据持续转化为训练数据，驱动模型迭代形成正向飞轮。

预览

十大功能模块

Data Flywheel：两条并行路径

批量路径（M6）：量产车每日路测数据经 COPY INTO 批量入湖，MERGE INTO 清洗后，通过

06_flywheel_scene_extract.sql

实时路径（M10）：量产车实时安全事件经 Kafka PIPE 秒级入湖，Dynamic Table 1 分钟刷新后，通过

04_flywheel_bridge.sql

两条路径汇入统一的

sim_ods_scenarios

四、ClickZetta Lakehouse 技术优势

1. SQL-first，零额外依赖

全部处理逻辑基于标准 SQL，无需 Python / Spark / Flink 环境。AI 推理（DTC 诊断、场景分类、标注质量评分）通过

AI_COMPLETE()

-- 在 SQL 中直接调用 LLM，逐行完成 DTC 故障诊断 UPDATE fleet_dwd_dtc_records SET ai_diagnosis = AI_COMPLETE( 'conn_dashscope:deepseek-v3', '诊断以下故障码，返回根因和建议：' || dtc_code ) WHERE ai_diagnosis IS NULL;

2. Dynamic Table 替代 Spark Streaming

传统方案需要独立维护 Flink/Spark Streaming 集群处理增量数据。Lakehouse Dynamic Table 用一条 SQL 声明即实现增量自动刷新：

CREATE DYNAMIC TABLE veh_dws_realtime_alert REFRESH INTERVAL 1 MINUTE -- 1 分钟级实时告警 AS SELECT vehicle_id, COUNT(*) AS event_count, MAX(risk_level) AS max_risk FROM veh_dwd_safety_events_clean WHERE msg_timestamp >= CURRENT_TIMESTAMP() - INTERVAL 5 MINUTE GROUP BY vehicle_id HAVING MAX(risk_level) IN ('L3','L4');

无需部署 Streaming 集群，无需管理 checkpoint，刷新间隔从 1 分钟到 6 小时灵活配置。

3. Kafka PIPE 实现准实时接入

原生支持 Kafka 持续消费，4 个 Topic 对应 4 个独立 PIPE，安全事件 PIPE 批次 10 秒，确保告警链路 <30 秒端到端延迟：

CREATE PIPE pipe_vehicle_safety_events BATCH_INTERVAL_IN_SECONDS = '10' -- 安全事件低延迟 AS COPY INTO veh_ods_safety_events FROM read_kafka(...);

4. INVERTED INDEX 支撑场景语义检索

仿真场景库百万级规模下，通过全文倒排索引实现毫秒级语义检索，无需独立的搜索引擎（Elasticsearch 等）：

CREATE INVERTED INDEX idx_scenario_desc ON TABLE sim_ods_scenarios (description, tags) PROPERTIES ('analyzer' = 'keyword'); -- 按语义检索场景 SELECT * FROM sim_ods_scenarios WHERE description LIKE '%雨天%' AND tags LIKE '%NEAR_MISS%';

5. 存算分离，按需扩缩

Lakehouse 存算分离架构，路测批量处理期间弹性扩展计算资源，空闲时自动缩减，相比固定集群方案节省 40%~70% 基础设施成本。

6. 统一治理，消除数据孤岛

ODS / DWD / DWS 三层全在一个 Lakehouse 实例中管理，标注数据、遥测数据、训练数据、评测数据共享统一的权限体系、数据血缘和 Time Travel（历史版本回溯）能力。

五、客户价值

效率提升

成本降低

• 基础设施：消除 Spark 集群 + Flink 集群 + 多套数据库的独立部署，降低 40%~60% 基础设施成本
• 运维人力：SQL-first 无需 Spark/Flink 工程师，数据工程师直接维护
• 模型迭代：数据闭环加速，每个模型迭代周期缩短，间接降低 GPU 训练成本

安全与合规

• L1-L4 风险事件全链路记录，证据链完整，满足 GB/T 40429 等自动驾驶相关法规要求
• OTA 灰度决策留有完整日志，可回溯每次升级的评测依据
• 安全事件与仿真场景双向关联，可追溯"该场景来自哪次量产事件"

竞争力提升

数据闭环周期从数周缩短到数天，意味着同等时间内可完成更多次模型迭代，在 NOA 城区功能等快速演进的赛道中建立数据飞轮优势。

六、方案注意事项

6.1 Lakehouse 建表兼容性

以下问题在实测中已全部遇到并解决，部署前需严格遵守：

PRIMARY KEY(event_id, trigger_ts)

(col1, col2)

change_tracking

DAYS(TIMESTAMP_MILLIS(ts_ms))

col['key']

col:key

6.2 Kafka PIPE 网络配置

最关键的前置条件：

read_kafka

SASL_PLAINTEXT

部署前必须确认：

✅ Lakehouse VCluster 与 Kafka 在同一 VPC，或已打通 VPC 对等连接 ✅ Kafka 安全组已放通 Lakehouse 节点 IP 段 → TCP 9092 入站 ✅ Kafka 提供了 SASL_PLAINTEXT 接入点（阿里云 Serverless 需手动添加 VPC 接入点） ✅ 用 read_kafka SELECT 连通性验证 SQL 测试通过（10 秒内返回数据）

如网络暂时无法打通，可使用 OSS 中转方案：

Kafka → 阿里云函数计算 FC（同 VPC）→ OSS → Lakehouse COPY INTO

6.3 数据规模与性能规划

6.4 AI_COMPLETE 成本控制

AI_COMPLETE

• 按需触发：只对

  WHERE ai_diagnosis IS NULL

  WHERE ai_diagnosis IS NULL
  的新记录调用，不重复推理
• 分级调用：规则层（Window Function）先筛选异常，只对命中规则的记录触发 AI，如交易异常检测仅对 L2+ 风险调用
• Prompt 简洁：封闭式输出（只返回标签名）比开放式生成 token 消耗少 5~10 倍
• 批量处理：Dynamic Table 批次刷新天然实现批量推理，而非逐条实时调用

6.5 Studio Task 调度建议

• M1 批量入湖建议设置

  AUTO_SUSPEND_IN_SECOND = 600

  AUTO_SUSPEND_IN_SECOND = 600
  ，避免 VCluster 频繁暂停导致 CZLH-60011
• Studio Task 中固化 Session Flag（

  set cz.sql.cast.string.to.json.as.parse=true

  set cz.sql.cast.string.to.json.as.parse=true
  等），每次连接都需重新设置
• PIPE 的

  RESET_KAFKA_GROUP_OFFSETS

  RESET_KAFKA_GROUP_OFFSETS
  仅在首次创建时生效，重建 PIPE 前记录当前 offset

七、验证状态（2026-06-05 实测）

00_create_schema.sql

10-vehicle-kafka-ingest/

ads_full_loop

八、快速部署步骤

前置条件

• ClickZetta Lakehouse workspace 已就绪
• 如使用 AI_COMPLETE，需预先

  CREATE CONNECTION

  CREATE CONNECTION
  （conn_dashscope 等）
• 如使用 Kafka PIPE，需确保 Lakehouse VCluster 与 Kafka 在同一 VPC（详见 §6.2）

Step 1：建表（选 A 或 B）

方式 A — 直接执行 SQL

# M1-M8 全部 27 张表（含 8 张 DWS Dynamic Table） run 00-setup/00_create_schema.sql # M10 实时链路（9 张表 + 4 个 Kafka PIPE 定义） run 10-vehicle-kafka-ingest/01_ods_tables.sql run 10-vehicle-kafka-ingest/02_dwd_tables.sql run 10-vehicle-kafka-ingest/03_dws_tables.sql

方式 B — 通过 Studio Task 执行

在 Lakehouse Studio 的

ads_full_loop

1. ads_01_ddl_ods

   ads_01_ddl_ods
   — ODS 层 10 张表

2. ads_02_ddl_dwd

   ads_02_ddl_dwd
   — DWD 层 15 张表

3. ads_03_ddl_dws

   ads_03_ddl_dws
   — DWS 层 8 张 Dynamic Table

4. ads_04_rt_ods_pipe

   ads_04_rt_ods_pipe
   — M10 实时链路 ODS + PIPE（需 Kafka VPC 连通）

5. ads_05_rt_dwd

   ads_05_rt_dwd
   — M10 DWD Dynamic Table

6. ads_06_rt_dws

   ads_06_rt_dws
   — M10 DWS Dynamic Table

Step 2：写入演示数据

run 09-full-loop-demo/01_sample_data_gen.sql

Step 3：运行全链路 Pipeline

run 09-full-loop-demo/02_run_full_loop.sql

Step 4：验证 Data Flywheel

run 09-full-loop-demo/03_validate_flywheel.sql

Step 5：部署定时调度

在 Studio 中执行

ads_offline_pipeline

0 2 * * *

九、Studio Task 清单

ads_01_ddl_ods

ads_02_ddl_dwd

ads_03_ddl_dws

ads_04_rt_ods_pipe

ads_05_rt_dwd

ads_06_rt_dws

ads_07_rt_flywheel

ads_offline_pipeline

相关文档

核心数据处理

查询加速与检索

AI 与 Python

安全与治理

任务调度与运维