OpenLake Studio 智能湖仓数据开发

第一步：Openlake—电商直播数据处理

首先，您将亲手构建一个现代化的数据湖仓体系。我们将引导您：

• 将RDS中的业务数据同步至开放数据湖（即数据湖构建DLF）。

• 通过MaxCompute、Serverless Spark、Serverless StarRocks、Hologres 多引擎协同工作。

• 完成明细数据构建、轻度汇总、加速查询和 AI 应用的一体化流程。

第二步：DataWorks Agent—自动构建数据工作流

本次动手实践的核心亮点——DataWorks Agent。我们将扮演ETL开发人员的角色，用自然语言下达指令（如“帮我构建直播订单分析工作流”），激活我们的 Agent。看它如何根据需求文档，自动完成需求分析、智能规划、代码生成、工作流编排乃至发布上线的全部工作。通过此环节，我们将体验到Agent为ETL工作带来的效率飞跃。

第三步：Notebook与ChatBI 

当 ETL 流程就绪后，我们进入数据消费的新范式：

• 使用专业的 Notebook 进行交互式探索。

• 像聊天一样通过 ChatBI 发起即时问答，获取关键业务洞察。

非DataWorks侧环境准备

• 数据湖构建DLF：开通DLF并创建数据目录。

• MaxCompute：创建MaxCompute项目并绑定DLF数据目录。

  重要

  目前创建MaxCompute外部项目，需提交工单联系MaxCompute产品团队创建。

• Hologres：开通Hologres并绑定DLF数据目录。

• EMR Serverless Spark：开通EMR Serverless Spark，创建工作空间并绑定DLF数据目录 。

• EMR Serverless StarRocks：开通Serverless StarRocks并绑定DLF数据目录。

DataWorks侧环境准备

1. 开通DataWorks，并创建OpenLake工作空间。

2. 绑定DLF数据目录。

   

3. 绑定上述计算资源：MaxCompute、Hologres、EMR Serverless Spark、EMR Serverless StarRocks。

4. 创建Serverless资源组。

   

5. 将资源组绑定到OpenLake工作空间。

DataWorks侧数据源准备

一、创建电商直播数据处理工作流

回到演练页，单击创建电商直播数据处理工作流并确认，系统将为您自动生成一个新的工作流容器。

二、同步 RDS 电商直播数据至 DLF

1. 单击同步RDS电商直播数据至DLF按钮，启动创建数据集成/离线同步节点流程。

2. 单击确认后，节点创建成功。

3. 鼠标移动至该节点上，单击打开节点进入配置页面。

   

4. 选择数据来源为MySQL，数据去向为DLF，并完成相关数据源及资源组的设置。

   选择我们在已创建好的RDS数据源和DLF数据源，以及绑定到当前工作空间的资源组，并保障资源组与数据源可联通。

   

5. 单击下一步，进行任务配置。

6. 在数据来源处，选择来源表为ctlive_trd_order_create。在数据去向处，单击一键生成目标表结构功能自动建表。

   可按需选择更多来源表。

   

7. 设置字段映射，默认按名称匹配即可。

8. 脏数据策略建议设置为不容忍脏数据。

9. 单击右侧工具栏的调试配置，设置预先准备好的计算资源及资源组。

10. 单击节点编辑页面上方的保存和运行按钮，开始数据同步。

三、通过MaxCompute SQL构建订单数据明细

1. 单击MaxCompute：构建订单数据明细按钮，创建 MaxCompute SQL 节点。

2. 单击确认后，节点创建成功。

3. 鼠标移动至该节点上，单击打开节点进入配置页面。系统自动填充如下业务处理代码。

   单击打开MaxCompute SQL代码。

   -- 订单明细表
   -- 作用：作为DWD层（明细层）表，用于存储标准化的订单明细数据，支持下游多维分析（如用户行为、商品销售趋势）。
   
   -- 第一部分：创建目标表
   -- 启用Schema模式，允许使用多级命名空间（如 e_commerceanalytics_mc_hzext.ctlive_dwd）。
   SET odps.namespace.schema=true;
   
   -- 删除已存在的同名表，避免重复创建导致的元数据冲突。
   drop table if exists ctlive_dwd_trd_order_create_di
   ;
   -- 创建分区表，字段: 包含订单基础信息（用户、商品、金额）、地理信息（省/市/区）、设备信息（终端类型、网络）等。
   -- 分区字段: ds（日期分区），用于按天管理数据，提升查询效率。
   create table if not exists e_commerceanalytics_mc_hzext.ctlive_dwd.ctlive_dwd_trd_order_create_di
   (
     date_key         string comment '日期键'
     ,live_room_id    string comment '直播间ID'
     ,visitor_id      string comment '访问用户ID'
     ,consumer_id     string comment '消费者ID'
     ,user_type       string comment '用户类型'
     ,anchor_id       string comment '主播ID'
     ,anchor_name     string comment '主播名称'
     ,agency_id       string comment '机构ID'
     ,agency_name     string comment '机构名称'
     ,seller_id       string comment '商家ID'
     ,seller_name     string comment '商家名称'
     ,shop_id         string comment '店铺ID'
     ,item_id         string comment '商品ID'
     ,leaf_cate_id    string comment '叶子类目ID'
     ,brand_id        string comment '品牌ID'
     ,brand_name      string comment '品牌名称'
     ,parent_order_id string comment '父订单ID'
     ,order_id        string comment '订单ID'
     ,order_status    string comment '订单状态'
     ,item_quantity   bigint comment '商品数量'
     ,order_amount    double comment '订单金额（元）'
     ,discount_amount double comment '优惠金额（元）'
     ,shipping_fee    double comment '运费（元）'
     ,total_amount    double comment '实付金额（元）'
     ,create_time     string comment '订单创建时间'
     ,pay_time        string comment '订单支付时间'
     ,ship_time       string comment '订单发货时间'
     ,complete_time   string comment '订单完成时间'
     ,cancel_time     string comment '订单取消时间'
     ,province_code   string comment '省份编码'
     ,province_name   string comment '省份名称'
     ,city_code       string comment '城市编码'
     ,city_name       string comment '城市名称'
     ,district_code   string comment '区县编码'
     ,district_name   string comment '区县名称'
     ,device_type     string comment '设备类型'
     ,os              string comment '操作系统'
     ,network_type    string comment '网络类型'
   )
   COMMENT '直播间订单创建明细表'
   PARTITIONED BY (
     ds STRING COMMENT '分区字段'
   );
   
   -- 第二部分：向目标表插入清洗后的数据
   -- 临时解除ODPS的全表扫描限制，允许无分区条件的查询（仅当前会话生效）。
   SET odps.sql.allow.fullscan = true;
   
   -- 使用 INSERT OVERWRITE 方式将清洗后的数据写入目标表，并按 date_key 字段自动分区
   INSERT OVERWRITE TABLE e_commerceanalytics_mc_hzext.ctlive_dwd.ctlive_dwd_trd_order_create_di PARTITION (ds)
   -- 从直播电商订单表中查询订单数据，并与地理维度表进行左连接，补充省、市、区县的名称信息。
   select  o.date_key ---日期键
           ,o.live_room_id ---直播房间ID
           ,o.visitor_id ---访客ID
           ,o.consumer_id ---消费者ID
           ,o.user_type ---用户类型
           ,o.anchor_id ---主播ID
           ,o.anchor_name ---主播名称
           ,o.agency_id ---代理ID
           ,o.agency_name ---代理名称
           ,o.seller_id ---卖家ID
           ,o.seller_name ---卖家名称
           ,o.shop_id ---店铺ID
           ,o.item_id ---商品ID
           ,o.leaf_cate_id ---叶子类目ID
           ,o.brand_id ---品牌ID
           ,o.brand_name ---品牌名称
           ,o.parent_order_id ---父订单ID
           ,o.order_id ---订单ID
           ,o.order_status ---订单状态
           ,o.item_quantity ---商品数量
           ,o.order_amount ---订单金额
           ,o.discount_amount ---折扣金额
           ,o.shipping_fee ---运费
           ,o.total_amount ---总金额
           ,o.create_time ---创建时间
           ,o.pay_time ---支付时间
           ,o.ship_time ---发货时间
           ,o.complete_time ---完成时间
           ,o.cancel_time ---取消时间
           ,coalesce(g.province_code,o.province_code) as province_code ---省份代码
           ,coalesce(g.province_name,o.province_name) as province_name ---省份名称
           ,coalesce(g.city_code,o.city_code) as city_code ---城市代码
           ,coalesce(g.city_name,o.city_name) as city_name ---城市名称
           ,coalesce(g.district_code,o.district_code) as district_code ---区县代码
           ,coalesce(g.district_name,o.district_name) as district_name ---区县名称
           ,o.device_type ---设备类型
           ,o.os ---操作系统
           ,o.network_type ---网络类型
           ,o.date_key ---插入分区
   from    e_commerceanalytics_mc_hzext.ctlive_ods.ods_ctlive_trd_order_create o
   left outer join e_commerceanalytics_mc_hzext.ctlive_dim.dim_pub_geography_df g
   on      o.province_code = g.province_code
   and     o.city_code = g.city_code
   and     o.district_code = g.district_code
   and     g.region_level = 4;
   
   -- 第三部分：数据验证
   -- 1. 数据总量验证
   -- 1.1 ODS层数据总量
   SET odps.namespace.schema=true;
   SET odps.sql.allow.fullscan = true;
   select count(*) from e_commerceanalytics_mc_hzext.ctlive_ods.ods_ctlive_trd_order_create;
   
   -- 1.2 DWD层数据总量
   SET odps.namespace.schema=true;
   SET odps.sql.allow.fullscan = true;
   select count(*) from e_commerceanalytics_mc_hzext.ctlive_dwd.ctlive_dwd_trd_order_create_di;
   
   -- 2. 数据分区验证
   -- 2.1 ODS层数据分区
   SET odps.namespace.schema=true;
   SET odps.sql.allow.fullscan = true;
   select date_key,count(*) from e_commerceanalytics_mc_hzext.ctlive_ods.ods_ctlive_trd_order_create group by date_key order by date_key desc limit 1000 ;
   
   -- 2.2 DWD层数据分区
   SET odps.namespace.schema=true;
   SET odps.sql.allow.fullscan = true;
   select ds,count(*) from e_commerceanalytics_mc_hzext.ctlive_dwd.ctlive_dwd_trd_order_create_di group by ds order by ds desc limit 1000;
   
   -- 3.明细数据查询
   SET odps.namespace.schema=true;
   SET odps.sql.allow.fullscan = true;
   select * from e_commerceanalytics_mc_hzext.ctlive_dwd.ctlive_dwd_trd_order_create_di where ds='20250618' limit 1000;

   

4. 单击右侧工具栏的调试配置，设置预先准备好的计算资源及资源组。

5. 单击节点编辑页面上方的运行按钮，运行任务。

6. 任务运行完成后，可在查询结果区域查看任务执行结果及订单明细数据。

   

四、通过Serverless Spark SQL构建订单汇总宽表

1. 回到实践方案的工作流页，单击Serverless Spark：构建订单汇总宽表，创建Serverless Spark SQL节点。

2. 单击确认后，节点创建成功。

   

3. 鼠标移动至该节点上，单击打开节点进入配置页面。系统自动填充如下业务处理代码。

   单击查看Serverless Spark SQL代码。

   -- 订单创建轻度汇总表
   -- 作用：作为轻度聚合层表，用于按主播、省份、用户类型等多维汇总订单数据，支持下游快速获取主播、省份等维度的支付GMV、订单量等核心指标
   
   -- 第一部分：创建目标表
   -- 删除已存在的同名表，避免重复创建导致的元数据冲突。
   drop table if exists ctlive_dws.dws_ctlive_trd_order_create_nd;
   
   -- 创建分区表，包含字段: order_id(订单ID), user_id(用户ID), product_id(商品ID), amount(金额), province(省), city(市), district(区), device_type(终端类型), network(网络类型)等。
   -- 分区字段: ds（日期分区），用于按天、主播、省份等维度管理数据，提升查询效率。
   create table ctlive_dws.dws_ctlive_trd_order_create_nd
   (
       date_key          string comment '日期键'
       ,anchor_id        string comment '主播ID'
       ,anchor_name      string comment '主播名称'
       ,agency_id        string comment '机构ID'
       ,agency_name      string comment '机构名称'
       ,province_name    string comment '省份名称'
       ,user_type        string comment '用户类型'
       ,seller_id        string comment '商家ID'
       ,seller_name      string comment '商家名称'
       ,shop_id          string comment '店铺ID'
       ,item_id          string comment '商品ID'
       ,leaf_cate_id     string comment '叶子类目ID'
       ,brand_id         string comment '品牌ID'
       ,brand_name       string comment '品牌名称'
       ,order_count      bigint comment '订单数'
       ,order_user_count bigint comment '下单人数'
       ,paid_order_count bigint comment '支付订单数'
       ,paid_user_count  bigint comment '支付人数'
       ,gmv              double comment 'GMV 订单金额总和'
       ,total_amount     double comment '实付总额'
       ,discount_amount  double comment '优惠总额'
       ,shipping_fee     double comment '运费总额'
       ,item_quantity    bigint comment '商品总件数'
   )
   partitioned by 
   (
       ds                string comment '分区字段'
   );
   
   -- 第二部分：向目标表插入清洗后的数据
   -- 使用 INSERT OVERWRITE 方式将清洗后的数据写入目标表
   INSERT OVERWRITE TABLE  ctlive_dws.dws_ctlive_trd_order_create_nd partition (ds)
   -- 从直播电商订单明细表中查询订单数据，按订单日期、主播、代理机构、省份、用户类型聚合统计订单量、GMV等核心指标，并按 date_key 字段自动分区
   select  date_key -- 日期键
           ,anchor_id -- 主播ID
           ,anchor_name -- 主播名称
           ,agency_id -- 代理ID
           ,agency_name -- 代理名称
           ,province_name -- 省份名称
           ,user_type -- 用户类型
           ,seller_id ---卖家ID
           ,seller_name ---卖家名称
           ,shop_id ---店铺ID
           ,item_id ---商品ID
           ,leaf_cate_id ---叶子类目ID
           ,brand_id ---品牌ID
           ,brand_name ---品牌名称
           ,count(order_id) as order_count -- 订单总数
           ,count(distinct consumer_id) as order_user_count -- 下单用户数
           ,count(case    when order_status in ('paid','shipped','completed') then order_id end) as paid_order_count -- 已支付订单数
           ,count(distinct case    when order_status in ('paid','shipped','completed') then consumer_id end) as paid_user_count -- 已支付用户数
           ,sum(order_amount) as gmv -- 商品交易总额
           ,sum(total_amount) as total_amount -- 总金额
           ,sum(discount_amount) as discount_amount -- 优惠金额
           ,sum(shipping_fee) as shipping_fee -- 运费
           ,sum(item_quantity) as item_quantity -- 商品数量
           ,date_key as ds -- 日期分区
   from    ctlive_dwd.dwd_ctlive_trd_order_create_di
   group by date_key
            ,anchor_id
            ,anchor_name
            ,agency_id
            ,agency_name
            ,province_name
            ,user_type
            ,seller_id
            ,seller_name
            ,shop_id
            ,item_id
            ,leaf_cate_id
            ,brand_id
            ,brand_name;
   
   -- 第三部分：数据验证
   -- 1. 从直播业务交易层按自然日汇总表中查询指定分区的数据
   select * from ctlive_dws.dws_ctlive_trd_order_create_nd where ds='20250618' limit 100;

   

4. 单击右侧工具栏的调试配置，设置预先准备好的计算资源及资源组。

   

5. 单击节点编辑页面上方的运行按钮，运行任务。

   建议只运行代码最后select语句。

6. 任务运行完成后，可在查询结果区域查看任务执行结果及订单汇总宽表数据。

   

五、Serverless StarRocks SQL：分析订单销售省份分布

1. 回到实践方案的工作流页，单击Serverless StarRocks：订单销售省份分布，创建 StarRocks SQL 节点 。

2. 单击确认后，节点创建成功。

   

3. 鼠标移动至该节点上，单击打开节点进入配置页面。系统填充如下业务处理代码。

   单击查看Serverless StarRocks SQL代码。

   -- 订单销售省份分布表
   -- 作用：作为ADS层（应用分析层）表，用于存储按省份维度统计的订单核心指标（总订单量、用户数、支付率、GMV等）。
   
   -- 第一部分：创建目标表
   -- 创建非分区表，包含字段: 省份、总订单数、下单用户数、支付用户数、支付转化率、总交易额、实际支付金额、客单价。
   CREATE TABLE dlf_catalog.ctlive_ads.ads_ctlive_trd_order_create_province_nd (
       province_name        STRING COMMENT '省份名称',
       total_orders         BIGINT COMMENT '总订单数',
       total_users          BIGINT COMMENT '下单用户数',
       paid_users           BIGINT COMMENT '支付用户数',
       conversion_rate      DOUBLE COMMENT '支付转化率(支付用户数/下单用户数)',
       gmv                  DOUBLE COMMENT '总交易额(GMV)',
       paid_gmv             DOUBLE COMMENT '实际支付金额',
       avg_order_value      DOUBLE COMMENT '客单价(支付GMV/支付用户数)'
   ) COMMENT 'ADS订单创建省份分布';
   
   -- 第二部分：向目标表插入清洗后的数据
   -- 使用 INSERT OVERWRITE 方式将清洗后的数据写入目标表
   INSERT OVERWRITE dlf_catalog.ctlive_ads.ads_ctlive_trd_order_create_province_nd
   -- 将直播电商订单数据按省份聚合统计核心指标，覆盖总订单量、用户数、支付转化率、GMV等维度，并按支付GMV降序排列
   SELECT
       province_name -- 省份名称
   ,
       SUM(order_count) AS total_orders -- 总订单量
   ,
       SUM(order_user_count) AS total_users -- 总用户数
   ,
       SUM(paid_user_count) AS paid_users -- 支付用户数
   ,
       IF(
           SUM(order_user_count) > 0,
           ROUND(
               CAST(SUM(paid_user_count) AS double) / SUM(order_user_count),
               4
           ),
           0.0
       ) AS conversion_rate -- 转化率
   ,
       ROUND(SUM(gmv)) AS gmv -- 总GMV
   ,
       ROUND(SUM(total_amount)) AS paid_gmv -- 支付GMV
   ,
       IF(
           SUM(paid_user_count) > 0,
           ROUND(SUM(total_amount) / SUM(paid_user_count), 2),
           0.0
       ) AS avg_order_value -- 客单价
   FROM
       dlf_catalog.ctlive_dws.dws_ctlive_trd_order_create_nd
   WHERE
       province_name IS NOT NULL
   GROUP BY
       province_name
   ORDER BY
       paid_gmv DESC;
   
   
   -- 第三部分：数据查询
   -- 1.从预聚合的省份粒度电商交易表中提取核心指标，并按支付GMV降序展示TOP10省份的交易表现
   SELECT 
       province_name AS 省份名称,
       total_orders AS 总订单数,
       total_users AS 总用户数,
       paid_users AS 支付用户数,
       conversion_rate AS 支付转化率,
       gmv AS 总GMV,
       paid_gmv AS 支付GMV,
       avg_order_value AS 平均客单价
   FROM
       dlf_catalog.ctlive_ads.ads_ctlive_trd_order_create_province_nd
   ORDER BY
       支付GMV DESC;

   

4. 单击右侧工具栏的调试配置，设置预先准备好的计算资源及资源组。

   

5. 单击节点编辑页面上方的运行按钮，运行任务。

6. 任务运行完成后，可在查询结果区域查看任务执行结果及订单明细数据。

六、Hologres：构建直播商品检索

1. 回到实践方案的工作流页，单击Hologres：构建直播商品检索，创建 Hologres SQL 节点。

2. 单击确认后，节点创建成功。

3. 鼠标移动至该节点上，单击打开节点进入配置页面。系统自动填充如下业务处理代码。

   单击查看Hologres SQL代码。

   -- 构建直播商品维度检索表
   CREATE TABLE IF NOT EXISTS tbl(id int);
   
   -- 从DLF中Load数据到Hologres中，并根据商品描述Embedding,创建向量索引
   CREATE DYNAMIC TABLE IF NOT EXISTS multimodal_search(
     CHECK(array_ndims(item_description_embeddings) = 1 AND array_length(item_description_embeddings, 1) =1024)
   )
   WITH (
       vectors = '{
         "item_description_embeddings": {
           "algorithm": "HGraph",
           "distance_method": "Cosine",
           "builder_params": {
           "base_quantization_type": "sq8_uniform",
           "max_degree": 64,
           "ef_construction": 400,
           "precise_quantization_type": "fp32",
           "use_reorder": true
           }
         }}',
       auto_refresh_mode='full',
       freshness='1 minutes'
   ) AS
   SELECT 
       item_id,
       item_name,
       item_subtitle,
       item_description,
       ai_embed ('qwen3_embedding_06b', item_description) AS item_description_embeddings,
       item_image_url
   FROM e_commerceanalytics_dlf.ctlive_ods.ods_ctlive_itm;
   
   -- 创建商品描述的全文索引
   CREATE INDEX IF NOT EXISTS idx_full_text_on_multimodal_search ON multimodal_search USING FULLTEXT (item_description);
   
   -- 查看刷新记录
   SELECT *
   FROM hologres.hg_dynamic_table_refresh_history
   WHERE dynamic_table_name ='multimodal_search'
   ORDER BY refresh_start DESC;
   
   -- 查看部分数据
   SELECT * FROM multimodal_search LIMIT 10;
   
   -- 使用全文索引检索，商品描述包含'高品质笔记本'的商品，召回BM25相似度最高的50个商品。
   SELECT
       *,
       text_search (item_description, '高品质笔记本') as bm25_score
   FROM
       multimodal_search
   WHERE
       text_search (item_description, '高品质笔记本') > 0
   ORDER BY
       text_search (item_description, '高品质笔记本') DESC
   LIMIT 50;
   
   -- 使用向量索引检索，商品描述包含'高品质笔记本'的商品，向量相似度最高的50个商品。
   SELECT
       *,
       approx_cosine_distance (item_description_embeddings, ai_embed ('qwen3_embedding_06b', '高品质笔记本')) as score
   FROM
       multimodal_search
   WHERE
       approx_cosine_distance (item_description_embeddings,ai_embed ('qwen3_embedding_06b', '高品质笔记本')) > 0
   ORDER BY
       approx_cosine_distance (item_description_embeddings, ai_embed ('qwen3_embedding_06b', '高品质笔记本')) DESC
   LIMIT 50;
   
   -- 同时使用向量索引检索和全文检索，商品描述包含'高品质笔记本'的商品，各自召回50个结果，然后使用Qwen3-4b模型做二次排序，最终从100个结果中取50个
   WITH
   -- 向量召回
   embedding_recall AS (
       SELECT
           item_id,
           item_name,
           item_subtitle,
           item_description,
           item_image_url
       FROM
           multimodal_search
       WHERE
           approx_cosine_distance (item_description_embeddings, ai_embed ('qwen3_embedding_06b', '高品质笔记本')) > 0
       ORDER BY
           approx_cosine_distance (item_description_embeddings, ai_embed ('qwen3_embedding_06b', '高品质笔记本')) DESC
       LIMIT 50
   ),
   -- 全文搜索召回（如果你不需要可以删除此 CTE）
   text_search_recall AS (
       SELECT
           item_id,
           item_name,
           item_subtitle,
           item_description,
           item_image_url
       FROM
           multimodal_search
       WHERE
           text_search (item_description, '高品质笔记本') > 0
       ORDER BY
           text_search (item_description, '高品质笔记本') DESC
       LIMIT 50
   ),
   -- 合并去重
   union_recall AS (
       SELECT
           item_id,
           item_name,
           item_subtitle,
           item_description,
           item_image_url
       FROM
           embedding_recall
       UNION
       SELECT
           item_id,
           item_name,
           item_subtitle,
           item_description,
           item_image_url
       FROM
           text_search_recall
   )
   SELECT
       *,
       ai_rank ('qwen3_4b', '高品质笔记本', item_description) AS score
   FROM
       union_recall
   ORDER BY
       score DESC
   LIMIT 50;

   

4. 单击右侧工具栏的调试配置，设置预先准备好的计算资源及资源组。

   

5. 单击节点编辑页面上方的运行按钮，运行任务。

6. 任务运行完成后，可在查询结果区域查看任务执行结果及构建直播商品检索结果。