大数据开发治理平台 DataWorks：DataWorks快速入门

步骤一：数据同步

1. 创建数据源。

   DataWorks通过创建数据源的方式，接入数据来源和数据去向，因此，本步骤需要分别创建MySQL和MaxCompute两个数据源。

   • MySQL数据源，用于连接数据来源（存储业务数据的MySQL数据库），为本教程提供原始业务数据。

     说明

     您无需准备本教程使用的原始业务数据，为方便测试和学习，DataWorks为您提供测试数据集，相关表数据已存储在公网MySQL数据库中，您只需创建MySQL数据源接入即可。

     创建MySQL数据源步骤

     1. 进入管理中心页面。

        登录DataWorks控制台，切换至目标地域后，单击左侧导航栏的更多 > 管理中心，在下拉框中选择对应工作空间后单击进入管理中心。

     2. 在左侧导航栏单击数据源 > 数据源列表，然后单击新增数据源，选择数据源类型为MySQL数据源。

     3. 配置MySQL数据源相关参数。

        说明

        • 表中未说明的参数保持默认值即可。

        • 首次新增数据源时，需要完成跨服务授权，请根据页面提示，授权服务关联角色AliyunDIDefaultRole。

        参数	描述
        数据源名称	本示例为MySQL_Source。
        配置模式	选择连接串模式。
        连接地址	单击新增地址。 主机地址IP：rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com 端口号：3306。 重要 本教程提供的数据仅作为阿里云大数据开发治理平台 DataWorks数据应用实操使用，所有数据均为测试数据，并且仅支持在数据集成模块读取数据。
        数据库名称	配置为retail_e_commerce。
        用户名	输入用户名workshop。
        密码	输入密码workshop#2017。

     4. 在连接配置区域，找到工作空间已绑定的资源组，单击连通状态列的测试连通性。

        说明

        如果MySQL数据源连通性测试失败，请检查是否为资源组绑定的VPC配置EIP，MySQL数据源需要资源组具备公网访问能力。

     5. 单击完成创建。

   • MaxCompute数据源，用于连接数据去向（MaxCompute数仓），将MaxCompute数据源绑定至数据开发后，能够为本教程提供数据存储和计算能力。

     • 如果您的项目空间中存在已添加的MaxCompute数据源，则无需进行此步骤。

     • 如果您的项目空间中没有添加MaxCompute数据源，在参考如下步骤创建。

       创建MaxCompute数据源步骤

       1. 在管理中心左侧导航栏单击数据源 > 数据源列表，然后单击新增数据源，选择数据源类型为MaxCompute数据源。

       2. 配置MaxCompute数据源相关参数。

          说明

          表中未说明的参数保持默认值即可。

          参数	描述
          数据源名称	本示例为MaxCompute_Source。
          地域	选择MaxCompute数据源所在地域。
          MaxCompute项目名称	选择创建MaxCompute项目。
          计算资源付费类型	选择按量付费。 说明 如果按量付费不可选，请单击按量付费后的去开通，完成MaxCompute服务的开通。
          默认Quota	下拉选择默认已存在的Quota。
          默认访问身份	阿里云主账号。

       3. 在连接配置区域，找到工作空间已绑定的资源组，单击连通状态列的测试连通性。

          说明

          MaxCompute项目名称全局唯一，如测试连通性不通过，报错提示为ErrorMessage:[ODPS-0420095: Access Denied - Authorization Failed [4002], You don't exist in project xxx.，是因为项目名称已存在，请修改项目名称后再试。

       4. 单击完成创建。

2. 数据开发绑定MaxCompute数据源。

   需要先将MaxCompute数据源绑定至数据开发，后续才能在数据开发模块中对MaxCompute的数据进行处理。

   1. 单击左上角图标，在弹出页面中单击全部产品 > 数据开发与运维 > DataStudio（数据开发）。

   2. 在左侧导航栏单击数据源（），找到已创建的MaxCompute数据源，单击绑定。

      说明

      如果您的数据开发模块已绑定了MaxCompute数据源，则无需进行此步骤。

      

3. 创建虚拟节点，用于统筹管理整个电商平台销售数据分析的业务流程。该节点为空跑任务，无须编辑代码。

   在左侧导航栏单击图标，进入数据开发页面，找到业务流程 > Workflow，然后右键Workflow，选择新建节点 > 通用 > 虚拟节点，自定义节点名称，本教程以Workshop为例。

4. 创建离线同步任务。

   本教程使用的测试数据涉及两张表（商品信息源表item_info和订单信息源表trade_order），这两张表存储于MySQL数据源关联的MySQL数据库中，本步骤需要分别创建两个离线同步节点（节点名称以ods_item_info和ods_trade_order为例），用于将item_info表和trade_order表同步至MaxCompute数据源关联的MaxCompute数仓中，然后再进行后续数据开发。

   1. 创建ods_item_info离线同步节点

      1. 右键Workflow，选择新建节点 > 数据集成 > 离线同步。

      2. 在新建节点页面，设置名称为ods_item_info。

      3. 配置数据来源为已创建的MySQL数据源，数据去向为已绑定至数据开发的MaxCompute数据源，中间的资源组选择工作空间已绑定的资源组，然后测试连通性都为可连通后，单击下一步。

         

      4. 配置数据来源和数据去向的相关参数。

         说明

         表中未说明的参数保持默认值即可。

         配置区域	参数	ods_item_info离线同步节点取值
         数据来源	表	item_info
         数据去向	表	单击一键生成目标表结构，在新建表窗口中，修改建表语句，将表名修改为ods_item_info，然后单击新建表。
         通道控制	脏数据策略	配置为不容忍脏数据。

      5. 单击离线同步任务配置窗口顶部的保存（），然后单击带参运行（）。

         重要

         在离线计算场景下，bizdate为业务交易发生的日期，也常被称为业务日期（business date）。例如，今天统计前一天的营业额，此处的前一天，指交易发生的日期，也就是业务日期。带参运行时，自定义参数取值保持默认（当前时间的前一天）即可。

         如果您在此时修改了bizdate取值，请在后续本文所有节点涉及带参运行时，同步修改bizdate取值。

   2. 创建ods_trade_order离线同步节点

      1. 右键Workflow，选择新建节点 > 数据集成 > 离线同步。

      2. 在新建节点页面，设置名称为ods_trade_order。

      3. 配置数据来源为已创建的MySQL数据源，数据去向为已绑定至数据开发的MaxCompute数据源，中间的资源组选择工作空间已绑定的资源组，然后测试连通性都为可连通后，单击下一步。

      4. 配置数据来源和数据去向的相关参数。

         说明

         表中未说明的参数保持默认值即可。

         配置区域	参数	ods_trade_order离线同步节点取值
         数据来源	表	trade_order
         数据去向	表	单击一键生成目标表结构，在新建表窗口中，修改建表语句，将表名修改为ods_trade_order，然后单击新建表。
         通道控制	脏数据策略	配置为不容忍脏数据。

      5. 单击离线同步任务配置窗口顶部的保存（），然后单击带参运行（）。

步骤二：数据清洗

数据已从MySQL同步至MaxCompute后，获得两张数据表（商品信息表ods_item_info和订单信息表ods_trade_order），您可以在DataWorks的数据开发模块对表中数据进行清洗、处理和分析，从而获取每日最畅销商品类目排名表。

1. 创建dim_item_info节点。

   基于ods_item_info表，处理商品维度数据，产出商品基础信息维度表dim_item_info。

   操作步骤

   1. 右键Workflow，选择新建节点 > MaxCompute > ODPS SQL，在新建节点页面，设置名称为dim_item_info后，单击确认，配置如下SQL。

      CREATE TABLE IF NOT EXISTS dim_item_info (
          gmt_modified                   STRING COMMENT '商品最后修改日期',
          gmt_create                     STRING COMMENT '商品创建时间',
          item_id                        BIGINT COMMENT '商品数字ID',
          title                          STRING COMMENT '商品标题',
          sub_title                      STRING COMMENT '商品子标题',
          pict_url                       STRING COMMENT '主图URL',
          desc_path                      STRING COMMENT '商品描述的路径',
          item_status                    BIGINT COMMENT '商品状态1：确认通过0：未确认通过',
          last_online_time               DATETIME COMMENT '最近一次开始销售时间，商品上架时间',
          last_offline_time              DATETIME COMMENT '销售结束时间，表示一个销售周期的结束，仅作用于拍卖商品',
          duration                       BIGINT COMMENT '有效期，销售周期，只有两个值，7天或14天',
          reserve_price                  DOUBLE COMMENT '当前价格',
          secure_trade_ordinary_post_fee DOUBLE COMMENT '平邮费用',
          secure_trade_fast_post_fee     DOUBLE COMMENT '快递费用',
          secure_trade_ems_post_fee      DOUBLE COMMENT 'EMS邮费',
          last_online_quantity           BIGINT COMMENT '商品最近一次上架时的库存数量',
          features                       STRING COMMENT '商品特征',
          cate_id                        BIGINT COMMENT '商品叶子类目ID',
          cate_name                      STRING COMMENT '商品叶子类目名称',
          commodity_id                   BIGINT COMMENT '品类ID',
          commodity_name                 STRING COMMENT '品类名称',
          is_virtual                     STRING COMMENT '是否虚拟商品',
          shop_id                        BIGINT COMMENT '商家ID',
          shop_nick                      STRING COMMENT '商家NICK',
          is_deleted                     BIGINT COMMENT '类目是否删除'
      )
      COMMENT '商品基础信息维度表'
      PARTITIONED BY (pt STRING COMMENT '业务日期, yyyymmdd')
      LIFECYCLE 365;
      
      
      -- 插入数据到 dim_item_info 表
      INSERT OVERWRITE TABLE dim_item_info PARTITION(pt='${bizdate}')
      SELECT
          gmt_create,
          gmt_modified,
          item_id,
          title,
          sub_title,
          pict_url,
          desc_path,
          item_status,
          last_online_time,
          last_offline_time,
          duration,
          cast(reserve_price as DOUBLE),
          cast(secure_trade_ordinary_post_fee as DOUBLE),
          cast(secure_trade_fast_post_fee as DOUBLE),
          cast(secure_trade_ems_post_fee as DOUBLE),
          last_online_quantity,
          features,
          cate_id,
          cate_name,
          commodity_id,
          commodity_name,
          is_virtual,
          shop_id,
          shop_nick,
          is_deleted
      FROM ods_item_info
      WHERE pt = '${bizdate}';

   2. 单击ODPS SQL配置窗口顶部的保存（），然后单击带参运行（）。

2. 创建dwd_trade_order节点。

   基于ods_trade_order表，对订单的详细交易数据进行初步清洗、转换和业务逻辑处理，产出交易下单明细事实表dwd_trade_order。

   操作步骤

   1. 右键Workflow，选择新建节点 > MaxCompute > ODPS SQL，在新建节点页面，设置名称为dwd_trade_order后，单击确认，配置如下SQL。

      CREATE TABLE IF NOT EXISTS dwd_trade_order (
          id               BIGINT COMMENT '主键，去重后的最新id',
          gmt_create       DATETIME COMMENT '创建时间',
          gmt_modified     DATETIME COMMENT '修改时间',
          sub_order_id     BIGINT COMMENT '子订单ID',
          parent_order_id  BIGINT COMMENT '父订单ID',
          buyer_id         BIGINT COMMENT '买家数字id',
          buyer_nick       STRING COMMENT '买家昵称，处理空值',
          item_id          BIGINT COMMENT '商品数字id',
          item_price       DECIMAL(38,18) COMMENT '商品价格，单位分',
          buy_amount       BIGINT COMMENT '购买数量',
          biz_type         BIGINT COMMENT '交易类型',
          memo             STRING COMMENT '备注，处理空值',
          pay_status       BIGINT COMMENT '支付状态',
          logistics_status BIGINT COMMENT '物流状态',
          status           BIGINT COMMENT '状态',
          seller_memo      STRING COMMENT '卖家的给交易的备注',
          buyer_memo       STRING COMMENT '买家给交易的备注',
          clean_ip         STRING COMMENT '清洗后的买家IP，过滤无效格式',
          end_time         DATETIME COMMENT '交易结束时间',
          pay_time         DATETIME COMMENT '付款的时间',
          is_sub           BIGINT COMMENT '是否是子订单1表示子订单',
          is_parent        BIGINT COMMENT '是否是父订单1表示父订单',
          shop_id          BIGINT COMMENT '商家id',
          total_fee        DECIMAL(38,18) COMMENT '去除折扣和调整后的子订单费用',
          is_large_order_flag BOOLEAN COMMENT '是否为大额订单标志'
      )
      COMMENT '交易下单明细事实表，包含初步清洗和业务逻辑处理'
      PARTITIONED BY (pt STRING COMMENT '业务日期, yyyymmdd')
      LIFECYCLE 365; -- 数据生命周期设置为365天
      
      
      INSERT OVERWRITE TABLE dwd_trade_order PARTITION(pt='${bizdate}')
      SELECT
          MAX(id) AS id, -- 假设使用最新的id作为去重标准
          gmt_create,
          gmt_modified,
          sub_order_id,
          parent_order_id,
          buyer_id,
          COALESCE(buyer_nick, '') AS buyer_nick, -- 处理buyer_nick为空的情况
          item_id,
          item_price,
          buy_amount,
          biz_type,
          COALESCE(memo, '') AS memo, -- 处理memo为空的情况
          pay_status,
          logistics_status,
          status,
          seller_memo,
          buyer_memo,
          CASE 
              WHEN ip LIKE '__.__.__.__' THEN NULL -- 过滤无效IP格式
              ELSE ip 
          END AS clean_ip,
          end_time,
          pay_time,
          is_sub,
          is_parent,
          shop_id,
          total_fee,
          CASE 
              WHEN total_fee >= 10000 THEN TRUE -- 假设大于10000分的订单为大额订单
              ELSE FALSE 
          END AS is_large_order_flag -- 添加业务逻辑标志
      FROM (
          SELECT
              *,
              ROW_NUMBER() OVER(PARTITION BY buyer_id, item_id, gmt_create ORDER BY id DESC) AS rn -- 用于去重的行号
          FROM ods_trade_order
          WHERE pt = '${bizdate}'
      ) AS sub_query
      WHERE rn = 1 -- 仅保留每个去重组的第一条记录
      GROUP BY 
          gmt_create,
          gmt_modified,
          sub_order_id,
          parent_order_id,
          buyer_id,
          buyer_nick,
          item_id,
          item_price,
          buy_amount,
          biz_type,
          memo,
          pay_status,
          logistics_status,
          status,
          seller_memo,
          buyer_memo,
          clean_ip,
          end_time,
          pay_time,
          is_sub,
          is_parent,
          shop_id,
          total_fee,
          is_large_order_flag;

   2. 单击ODPS SQL配置窗口顶部的保存（），然后单击带参运行（）。

3. 创建dws_daily_category_sales节点。

   基于dwd_trade_order表和dim_item_info表，对DWD层经过清洗和标准化的明细数据进行汇总，产出每日商品类目销售汇总表dws_daily_category_sales。

   操作步骤

   1. 右键Workflow，选择新建节点 > MaxCompute > ODPS SQL，在新建节点页面，设置名称为dws_daily_category_sales后，单击确认，配置如下SQL。

      CREATE TABLE IF NOT EXISTS dws_daily_category_sales (
          cate_id             BIGINT COMMENT '商品叶子类目ID',
          cate_name           STRING COMMENT '商品叶子类目名称',
          total_sales_amount  DECIMAL(38,18) COMMENT '商品类目总销售额，单位分',
          order_count         BIGINT COMMENT '订单数量'
      )
      COMMENT '每日商品类目销售汇总表'
      PARTITIONED BY (pt STRING COMMENT '业务日期, yyyymmdd')
      LIFECYCLE 365;
      
      
      INSERT OVERWRITE TABLE dws_daily_category_sales PARTITION(pt='${bizdate}')
      SELECT
          i.cate_id,
          i.cate_name,
          SUM(t.total_fee) AS total_sales_amount,
          COUNT(DISTINCT t.id) AS order_count
      FROM dwd_trade_order t
      JOIN dim_item_info i ON t.item_id = i.item_id AND t.pt = i.pt
      WHERE t.pt = '${bizdate}'
      GROUP BY t.pt, i.cate_id, i.cate_name;

   2. 单击ODPS SQL配置窗口顶部的保存（），然后单击带参运行（）。

4. 创建ads_top_selling_categories节点。

   基于dws_daily_category_sales表，产出每日最畅销商品类目排名表ads_top_selling_categories。

   操作步骤

   1. 右键Workflow，选择新建节点 > MaxCompute > ODPS SQL，在新建节点页面，设置名称为ads_top_selling_categories后，单击确认，配置如下SQL。

      CREATE TABLE IF NOT EXISTS ads_top_selling_categories (
          rank                BIGINT COMMENT '销量排名',
          cate_id             BIGINT COMMENT '商品叶子类目ID',
          cate_name           STRING COMMENT '商品叶子类目名称',
          total_sales_amount  DECIMAL(38,18) COMMENT '商品类目总销售额，单位分',
          order_count         BIGINT COMMENT '订单数量'
      )
      COMMENT '每日最畅销商品类目排名表'
      PARTITIONED BY (pt STRING COMMENT '业务日期, yyyymmdd');
      
      
      INSERT OVERWRITE TABLE ads_top_selling_categories PARTITION(pt='${bizdate}')
      SELECT
          rank,
          cate_id,
          cate_name,
          total_sales_amount,
          order_count
      FROM (
          SELECT
              DENSE_RANK() OVER(ORDER BY total_sales_amount DESC) AS rank,
              cate_id,
              cate_name,
              total_sales_amount,
              order_count
          FROM (
              SELECT
                  cate_id,
                  cate_name,
                  SUM(total_sales_amount) AS total_sales_amount,
                  SUM(order_count) AS order_count
              FROM dws_daily_category_sales
              WHERE pt = '${bizdate}'
              GROUP BY cate_id, cate_name
          ) agg_sub
      ) agg_outer
      WHERE rank <= 10;

   2. 单击ODPS SQL配置窗口顶部的保存（），然后单击带参运行（）。

步骤三：数据展示

您已经将从MySQL中获取的原始测试数据，经过数据开发处理，汇总于表ads_top_selling_categories中，现在可查询表数据，查看数据分析后的结果。

1. 单击左上角图标，在弹出页面中单击全部产品 > 数据分析 > SQL查询。

2. 在我的文件后单击 > 新建文件，自定义文件名后单击确定。

   

3. 在SQL查询页面，配置如下SQL。

   SELECT * FROM ads_top_selling_categories WHERE pt=${bizdate};

4. 单击顶部的运行（），根据页面提示，在右上角选择MaxCompute数据源后单击确定，然后在成本预估页面，单击运行。

5. 在查询结果中单击，查看可视化图表结果，您可以单击图表右上角的自定义图表样式。

   

6. 您也可以单击图表右上角保存，将图表保存为卡片，然后在左侧导航栏单击卡片（）查看。

   

步骤四：周期性调度

通过完成前文操作步骤，您已经获取了前一天各类商品的销售数据，但是，如果需要每天获取最新的销售数据，则可以为数据开发中各任务节点配置周期任务，使其周期性定时执行。

1. 单击左上角图标，在弹出页面中单击全部产品 > 数据开发与运维 > DataStudio（数据开发）。

2. 右键单击业务流程Workflow，单击看板进入画布，在画布中移动各节点位置并按下图拖拽出各节点的上下游依赖关系。

   

3. 单击右侧流程参数，配置参数名称为bizdate，参数值或表达式为$bizdate，单击保存。

   

4. 双击虚拟节点（Workshop），配置如下周期调度参数后，单击顶部的保存（）。

   说明

   其他参数保持默认即可。

   

5. 切换至Workflow业务流程页签，单击顶部的运行，确认运行业务流程，参数bizdate填写为前一天（例如今天为20240731，则此处填写为20240730），单击确定测试所有流程是否均能成功运行。

   

6. 所有节点均能成功运行后，点击顶部的提交，将流程中所有节点提交至运维中心。

   

7. 单击左上角图标，在弹出页面单击全部产品 > 数据开发与运维 > 运维中心（工作流）。

8. 在周期任务运维 > 周期任务中即可看到已创建的周期任务。

   说明

   如需展示如下图的所有上下游依赖节点，请右键单击Workshop节点，选择展开子节点 > 四层。