数据质量规则配置实践教程

前提条件

已购买数据质量功能。

数据准备

• 创建一个dataphin_v50_dev项目，用于数据校验及创建质量规则。

• 已发布用户表（user）、订单表（order1）和订单状态表（order_status_history）。

• 脚本语句如下，可下载样例数据，使用本地文件输入组件和MaxCompute输出组件将数据同步至Dataphin。

  样例数据：用户表.csv、订单表.csv、订单状态表.csv

  MaxCompute一键建表语句如下：

  订单表

  create table IF NOT EXISTS  `order1` (
          `customer_id` string comment '用户ID',
  	`crt_ord_id` double comment 'order_id',
  	`delivery_address_id` double comment '收货地址',
  	`buyer_id` double comment '买家ID',
  	`gmt_create` string comment '创建时间',
  	`gmt_modified` string comment '修改时间',
  	`order_time` string comment '下单时间',
  	`pay_time` string comment '支付时间',
  	`ship_time` string comment '发货时间',
  	`end_time` string comment '交易完成/交易取消 时间',
  	`total_items_amount` double comment '总商品金额，单价*件数',
  	`discount_amount` double comment '总折扣金额，单位分',
  	`delivery_amount` double comment '运费单位分',
  	`total_amount` double comment '实付总金额单位分',
  	`status` double comment '状态：-1已取消1新建2已支付3已发货4确认发货'
  )
  comment 'order1'
  lifecycle 3600

  用户表

  create table IF NOT EXISTS `user` (
  	`user_id` string comment '用户ID',
  	`name` string comment '姓名',
  	`gender` string comment '性别',
  	`age` string comment '年龄',
  	`phone_num` string comment '手机号码'
  )
  comment 'user'
  lifecycle 3600

  订单状态表

  CREATE TABLE IF NOT EXISTS order_status_history (
  	id Double NOT NULL COMMENT '自增主键',
  	order_id STRING COMMENT '关联的订单ID（外键，指向订单主表）',
  	`status` STRING COMMENT '订单状态（如created/paid/shipped/cancelled等）',
  	update_time STRING COMMENT '状态变更时间（通常按此字段排序）',
  	operator STRING COMMENT '操作人（可能是系统自动触发，如system，或人工操作，如user:1001）',
  	remark STRING COMMENT '变更备注（例如取消原因，“用户主动取消”）'
  )
  STORED AS ALIORC
  TBLPROPERTIES (
  	'columnar.nested.type' = 'true',
  	'transactional' = 'true'
  );

规则强度说明

• 强规则：当质量规则校验的结果为异常时，系统会报警并阻塞下游任务节点。

• 弱规则：当质量规则校验的结果为异常时，系统会报警但不阻塞下游任务节点。

质量规则配置流程

建议先创建调度配置，告警配置和异常归档表，便于顺利创建质量规则。、和的创建方式与空值质量规则一致，请参见。

基础场景配置示例

稍复杂场景配置示例

复杂场景配置示例

场景描述：发货表的订单支付状态流转有效性校验，当符合下方其中一个规则，则质量校验不会通过。

• 订单状态不能从已发货直接变为已取消。

• 订单必须先已支付才能变为已发货。

• 已退款的订单必须先有支付记录。

• 已签收的订单必须先有发货记录。

创建质量规则

1. 在Dataphin首页的顶部菜单栏，选择治理 > 数据质量。

2. 单击左侧导航栏的质量规则，在Dataphin表页签，单击添加监控对象按钮，进入添加监控对象对话框。

3. 在添加监控对象对话框中，配置参数后，单击配置质量规则。

   参数	描述
   板块/项目	选择项目dataphin_v50_dev。
   数据表	选择order_status_history。
   质量负责人	默认SuperAdmin。
   质量分权重	默认1。

4. 在质量规则详情页面，单击新建质量规则按钮，选择自定义配置，在新建质量规则对话框中，配置质量规则。

   参数		描述
   基本信息	规则名称	输入电商平台订单支付状态流转有效性校验。
   	规则强度	选择强规则。
   	描述	可以填写对质量规则的简单描述。例如： 订单状态不能从已取消直接变为已发货。 订单必须先已支付才能变为已发货。 已退款的订单必须先有支付记录。 已签收的订单必须先有发货记录。
   	配置方式	选择模板创建。
   	规则模板	选择自定义/自定义数据详情校验。
   	规则类型	选择有效性。
   模板配置	参数配置	根据规则模板默认展示，无需填写。
   规则配置	总行数SQL	输入如下SQL。 select count(*) from order_status_history；
   	异常行数SQL/异常数据SQL	输入SQL，详见表格下方SQL语句。
   校验配置	规则校验	选择异常率<=0，只要有一条异常数据，则校验不通过。
   归档配置	异常归档	开启异常归档。
   	归档模式	选择归档完整记录。
   	归档位置	选择自定义异常归档表（dataphin_v50_dev.order_status_history_exception_data）。
   调度属性配置	调度方式	选择定时调度。
   质量分配置	计分方式	默认质量校验状态。
   	质量分权重	默认为1。

   异常行数SQL

   SELECT 
   count(*)
   FROM 
   (    SELECT 
           id,
           order_id,
           status AS current_status,
           update_time AS current_update_time,
           operator AS current_operator,
           remark AS current_remark,
           -- 获取前一行数据
           LAG(id) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_id,
           LAG(status) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_status,
           LAG(update_time) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_update_time,
           LAG(operator) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_operator,
           LAG(remark) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_remark
       FROM order_status_history) temp_table
   WHERE (previous_status = 'cancelled' AND current_status = 'shipped')
      OR (previous_status <> 'paid' AND current_status = 'shipped')
      OR (previous_status = 'refunded' AND current_status NOT IN ('created', 'cancelled'))
      OR (previous_status = 'delivered' AND current_status = 'shipped')

   异常数据SQL

   SELECT 
       previous_id as  id,
       order_id,
       previous_status AS status,
       previous_update_time AS update_time,
       previous_operator AS operator,
       previous_remark AS remark
   FROM 
   (    SELECT
           id,
           order_id,
           status AS current_status,
           update_time AS current_update_time,
           operator AS current_operator,
           remark AS current_remark,
           -- 获取前一行数据
           LAG(id) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_id,
           LAG(status) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_status,
           LAG(update_time) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_update_time,
           LAG(operator) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_operator,
           LAG(remark) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_remark
       FROM order_status_history) temp_table
   WHERE (previous_status = 'cancelled' AND current_status = 'shipped')
      OR (previous_status <> 'paid' AND current_status = 'shipped')
      OR (previous_status = 'refunded' AND current_status NOT IN ('created', 'cancelled'))
      OR (previous_status = 'delivered' AND current_status = 'shipped')
   UNION ALL
   SELECT 
       id AS id,
       order_id,
       current_status AS status,
       current_update_time AS update_time,
       current_operator AS operator,
       current_remark AS remark
   FROM (    SELECT 
           id,
           order_id,
           status AS current_status,
           update_time AS current_update_time,
           operator AS current_operator,
           remark AS current_remark,
           -- 获取前一行数据
           LAG(id) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_id,
           LAG(status) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_status,
           LAG(update_time) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_update_time,
           LAG(operator) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_operator,
           LAG(remark) OVER (PARTITION BY order_id ORDER BY cast(update_time as TIMESTAMP )) AS previous_remark
       FROM order_status_history) temp_table
   WHERE (previous_status = 'cancelled' AND current_status = 'shipped')
      OR (previous_status <> 'paid' AND current_status = 'shipped')
      OR (previous_status = 'refunded' AND current_status NOT IN ('created', 'cancelled'))
      OR (previous_status = 'delivered' AND current_status = 'shipped')
   ORDER BY order_id, update_time;

5. 单击确定，完成空值校验质量规则创建。

6. 在规则配置列表，选择dataphin_v50_dev.order_status_history对象，单击操作列下的更多，选择试跑，在试跑对话框中，选择自定义校验范围，单击确定。

   试跑结束后，因检测到发货表中有订单流转状态存在异常，故试跑不通过。

   

7. 异常数据都记录在dataphin_v50_dev.order_status_history_exception_data异常归档表中，可在研发 > 数据研发 > 计算任务，创建MAX_COMPUTE_SQL任务查看异常数据。

   select  *
   from    dataphin_v50_dev.order_status_history_exception_data
   where   dataphin_quality_validate_date = '20251121'