EMR Serverless StarRocks AI Function构建智能驾驶训练数据管理平台

SET CATALOG default_catalog;
CREATE DATABASE IF NOT EXISTS ad_data_mining;
CREATE OBJECT TABLE ad_data_mining.camfront_objects PROPERTIES (
  "path" = "oss://<BUCKET>/CAM_FRONT/",
  "aliyun.oss.endpoint" = "oss-cn-hangzhou-internal.aliyuncs.com",
  "aliyun.oss.public_endpoint" = "oss-cn-hangzhou.aliyuncs.com",
  "aliyun.oss.access_key" = "<AK>",
  "aliyun.oss.secret_key" = "<SK>",
  "recursive" = "true",
  "file_pattern" = ".*\\.(jpg|jpeg|png)$"
);
REFRESH OBJECT TABLE ad_data_mining.camfront_objects;

CREATE TABLE dlf_catalog_demo.`default`.camfront_ai_enriched (
  id STRING, object_uri STRING, etag STRING, file_name STRING,
  caption STRING,            -- 场景自然语言描述
  metadata STRING,           -- 结构化元数据 JSON（天气/光照/路况/目标物/风险）
  vector ARRAY<FLOAT>,       -- 图像向量，2560 维
  weather STRING, lighting STRING, road_type STRING,
  is_cloudy_candidate INT, is_daytime INT, is_nighttime INT,
  scene_type STRING,
  dataset_tag INT            -- 检索圈选后回写：0 未选 / 1 阴天 / 2 阴天+行人
) ENGINE = paimon
PROPERTIES ("file.format"="parquet","primary-key"="id","merge-engine"="partial-update","bucket"="1");

INSERT INTO dlf_catalog_demo.`default`.camfront_ai_enriched
  (id, object_uri, etag, file_name, caption, metadata, vector,
   weather, lighting, road_type, is_cloudy_candidate, is_daytime, is_nighttime, scene_type, dataset_tag)
WITH enriched AS (
  SELECT o.object_uri, o.etag,
    regexp_extract(o.object_uri, '.*/(.+)$', 1) AS file_name,
    -- 场景描述（自由文本，4 参数带 model）
    ai_complete('qwen3.6-plus',
      'You are an autonomous driving data analysis assistant. Describe the road scene in 1-2 English sentences, highlighting weather, lighting, road type, key traffic participants and potential risks. Output only the descriptive text.',
      o.file, 'image') AS caption,
    -- 结构化元数据（4 参数带 schema：prompt, file, modality, PARSE_JSON —— 不带 model）
    ai_complete(
      'You are an autonomous driving data analysis assistant. Output structured metadata.',
      o.file, 'image',
      PARSE_JSON('{"type":"object","required":["weather","lighting","road_type","objects","risks"],"properties":{"weather":{"type":"string","enum":["sunny","cloudy","rainy","snowy","foggy","other"]},"lighting":{"type":"string","enum":["daytime","nighttime","dusk","tunnel","other"]},"road_type":{"type":"string","enum":["urban","expressway","rural","intersection","ramp","parking_lot","other"]},"objects":{"type":"array","items":{"type":"string"}},"risks":{"type":"array","items":{"type":"string"}}},"additionalProperties":false}')
    ) AS metadata,
    ai_embed_multimodal(o.file, 'image') AS vector
  FROM `default_catalog`.`ad_data_mining`.`camfront_objects` o
  LEFT ANTI JOIN dlf_catalog_demo.`default`.camfront_ai_enriched e
    ON o.object_uri = e.object_uri AND o.etag = e.etag
  WHERE object_file_is_image(o.file)
)
SELECT
  uuid(), object_uri, etag, file_name, caption, metadata, vector,
  get_json_string(metadata, '$.validated_result.weather')   AS weather,
  get_json_string(metadata, '$.validated_result.lighting')  AS lighting,
  get_json_string(metadata, '$.validated_result.road_type') AS road_type,
  IF(get_json_string(metadata, '$.validated_result.weather')  = 'cloudy',   1, 0),
  IF(get_json_string(metadata, '$.validated_result.lighting') = 'daytime',  1, 0),
  IF(get_json_string(metadata, '$.validated_result.lighting') = 'nighttime',1, 0),
  concat_ws('_', get_json_string(metadata,'$.validated_result.lighting'), get_json_string(metadata,'$.validated_result.weather')),
  0
FROM enriched;

-- 训练集均衡性评估：各天气占比，判断是否需要定向采集
SELECT weather, COUNT(*) AS cnt,
       ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 1) AS pct
FROM dlf_catalog_demo.`default`.camfront_ai_enriched
GROUP BY weather ORDER BY cnt DESC;
-- 混合检索：阴天 + 行人 高危 Corner Case（标量过滤 + 向量相似度 一条 SQL）
WITH seed AS (
  SELECT vector AS seed_vec FROM dlf_catalog_demo.`default`.camfront_ai_enriched
  WHERE is_cloudy_candidate = 1 ORDER BY id LIMIT 1
)
SELECT t.object_uri, t.caption, t.weather, t.road_type,
       json_query(t.metadata, '$.validated_result.objects') AS objects,
       cosine_similarity(t.vector, seed.seed_vec) AS similarity
FROM dlf_catalog_demo.`default`.camfront_ai_enriched t CROSS JOIN seed
WHERE t.weather = 'cloudy'
  AND t.metadata LIKE '%pedestrian%'
  AND cosine_similarity(t.vector, seed.seed_vec) > 0.8
ORDER BY similarity DESC;

-- 打标：阴天+行人(相似度>0.7)→2；仅阴天(相似度>0.4)→1；其余→0
CREATE TABLE dlf_catalog_demo.`default`.dataset_search_tags (id VARCHAR(65533), dataset_type INT);
INSERT INTO dlf_catalog_demo.`default`.dataset_search_tags
SELECT id,
  CASE
    WHEN id IN (SELECT t.id FROM dlf_catalog_demo.`default`.camfront_ai_enriched t,
                 (SELECT vector AS v FROM dlf_catalog_demo.`default`.camfront_ai_enriched WHERE is_cloudy_candidate=1 ORDER BY id LIMIT 1) s
                WHERE t.weather='cloudy' AND t.metadata LIKE '%pedestrian%' AND cosine_similarity(t.vector,s.v)>0.7) THEN 2
    WHEN id IN (SELECT t.id FROM dlf_catalog_demo.`default`.camfront_ai_enriched t,
                 (SELECT vector AS v FROM dlf_catalog_demo.`default`.camfront_ai_enriched WHERE is_cloudy_candidate=1 ORDER BY id LIMIT 1) s
                WHERE t.weather='cloudy' AND cosine_similarity(t.vector,s.v)>0.4) THEN 1
    ELSE 0 END
FROM dlf_catalog_demo.`default`.camfront_ai_enriched;
-- Partial Update 回写：只更新 dataset_tag，caption/metadata/vector 不动
INSERT INTO dlf_catalog_demo.`default`.camfront_ai_enriched
SELECT t.id, t.object_uri, t.etag, t.file_name, t.caption, t.metadata, t.vector,
  t.weather, t.lighting, t.road_type, t.is_cloudy_candidate, t.is_daytime, t.is_nighttime, t.scene_type,
  COALESCE(s.dataset_type, t.dataset_tag) AS dataset_tag
FROM dlf_catalog_demo.`default`.camfront_ai_enriched t
LEFT JOIN dlf_catalog_demo.`default`.dataset_search_tags s ON t.id = s.id;
-- 下游训练直接按标签取数
SELECT object_uri FROM dlf_catalog_demo.`default`.camfront_ai_enriched WHERE dataset_tag = 2;