数据质量校验任务：Apache DolphinScheduler内置验证器使用指南

1. 数据质量校验的痛点与解决方案

在数据驱动决策的时代，企业面临着日益严峻的数据质量挑战。据Gartner统计，数据质量问题导致企业平均每年损失1500万美元。Apache DolphinScheduler（海豚调度器）作为现代数据编排平台，提供了内置的数据质量（Data Quality）校验任务模块，帮助用户在数据处理流程中实现自动化的数据质量监控。

本文将系统介绍DolphinScheduler数据质量校验任务的设计原理、核心功能及实战应用，读完后您将能够：

• 理解数据质量校验任务的架构设计与工作流程
• 掌握6种内置验证器的配置与使用方法
• 实现数据质量校验与工作流的无缝集成
• 构建自定义数据质量监控规则与告警机制

2. 数据质量校验任务架构设计

2.1 核心组件架构

DolphinScheduler的数据质量校验任务基于Spark引擎实现，采用插件化架构设计，主要包含以下核心组件：

classDiagram
    class DataQualityTask {
        +init()
        +getScript()
        +operateInputParameter()
    }
    class DataQualityParameters {
        -ruleId: int
        -ruleInputParameter: Map
        -sparkParameters: SparkParameters
        +checkParameters(): boolean
    }
    class DataQualityTaskExecutionContext {
        -ruleId: int
        -ruleType: int
        -ruleName: String
        -hdfsPath: String
    }
    class RuleManager {
        +generateDataQualityParameter(): DataQualityConfiguration
    }
    class DataQualityConfiguration {
        -source: ConnectorConfiguration
        -target: ConnectorConfiguration
        -comparison: ComparisonConfiguration
    }
    
    DataQualityTask --> DataQualityParameters
    DataQualityTask --> DataQualityTaskExecutionContext
    DataQualityTask --> RuleManager
    RuleManager --> DataQualityConfiguration

2.2 工作流程

数据质量校验任务的执行流程可分为以下五个阶段：

flowchart TD
    A[任务初始化] --> B[参数解析与验证]
    B --> C[规则配置生成]
    C --> D[Spark命令构建]
    D --> E[任务提交与监控]
    E --> F[结果处理与告警]
    
    subgraph 初始化阶段
    A1[解析DataQualityParameters]
    A2[验证规则参数完整性]
    A3[设置Spark执行上下文]
    end
    
    subgraph 参数处理
    B1[补充系统参数]
    B2[处理正则表达式转义]
    B3[设置错误数据输出路径]
    end

关键实现代码位于DataQualityTask.java的init()方法中，该方法完成参数解析、规则配置生成及Spark命令构建：

public void init() {
    // 解析数据质量参数
    dataQualityParameters = JSONUtils.parseObject(dqTaskExecutionContext.getTaskParams(), DataQualityParameters.class);
    
    // 参数合法性校验
    if (!dataQualityParameters.checkParameters()) {
        throw new RuntimeException("data quality task params is not valid");
    }
    
    // 补充系统参数
    operateInputParameter(inputParameter, dataQualityTaskExecutionContext);
    
    // 生成数据质量配置
    RuleManager ruleManager = new RuleManager(inputParameter, dataQualityTaskExecutionContext);
    DataQualityConfiguration dataQualityConfiguration = ruleManager.generateDataQualityParameter();
    
    // 设置Spark应用参数
    dataQualityParameters.getSparkParameters().setMainArgs("\"" + 
        replaceDoubleBrackets(StringEscapeUtils.escapeJava(JSONUtils.toJsonString(dataQualityConfiguration))) + "\"");
}

3. 核心验证器类型与使用场景

DolphinScheduler提供了丰富的内置数据质量验证器，覆盖数据完整性、准确性、一致性等多个维度。根据DataQualityParameters.java的定义，系统通过ruleId标识不同的验证规则类型：

3.1 验证器类型与参数

验证器类型	ruleId	适用场景	核心参数	失败阈值
非空校验	1	关键字段非空验证	column、threshold	非空记录数 < 阈值
唯一性校验	2	主键/唯一索引验证	column、maxDuplicates	重复记录数 > 0
范围校验	3	数值字段范围限制	column、min、max	超出范围记录 > 阈值
正则表达式校验	4	格式验证（手机号、邮箱等）	column、regexp	不匹配记录数 > 阈值
表对比校验	5	数据同步一致性验证	sourceTable、targetTable、keyColumn	不一致记录数 > 阈值
空值占比校验	6	字段空值率监控	column、maxNullRatio	空值占比 > maxNullRatio

3.2 数据源支持

数据质量校验任务支持多种数据源，通过SRC_CONNECTOR_TYPE和TARGET_CONNECTOR_TYPE参数指定，包括：

// TaskExecutionContextFactory.java中定义的数据源类型常量
import static org.apache.dolphinscheduler.plugin.task.api.utils.DataQualityConstants.SRC_CONNECTOR_TYPE;
import static org.apache.dolphinscheduler.plugin.task.api.utils.DataQualityConstants.TARGET_CONNECTOR_TYPE;

支持的数据源类型包括：MySQL、PostgreSQL、Oracle、SQL Server、Hive、ClickHouse、Doris、StarRocks等，具体可参考dolphinscheduler-datasource-plugin模块下的实现。

4. 任务配置与实战示例

4.1 基础配置结构

数据质量校验任务的配置参数遵循JSON格式，由ruleId、ruleInputParameter和sparkParameters三部分组成：

{
  "ruleId": 4,
  "ruleInputParameter": {
    "column": "email",
    "regexp": "^[A-Za-z0-9+_.-]+@[A-Za-z0-9.-]+$",
    "threshold": "0"
  },
  "sparkParameters": {
    "mainClass": "org.apache.dolphinscheduler.data.quality.DataQualityApplication",
    "master": "yarn",
    "deployMode": "cluster",
    "driverCores": 1,
    "driverMemory": "1g",
    "executorCores": 2,
    "executorMemory": "2g",
    "numExecutors": 2
  }
}

4.2 正则表达式校验实战

以下是一个验证用户表邮箱格式的完整示例，使用正则表达式验证器（ruleId=4）：

4.2.1 任务配置

1. 基本信息：

   • 任务名称：user_email_format_check
   • 任务类型：DataQuality
   • 数据源：user_db（MySQL）
   • 目标表：t_user

2. 规则参数：

{
  "ruleId": 4,
  "ruleInputParameter": {
    "column": "email",
    "regexp": "^[A-Za-z0-9+_.-]+@[A-Za-z0-9.-]+$",
    "threshold": "0",
    "data_time": "${system.biz.date}"
  },
  "sparkParameters": {
    "master": "yarn",
    "deployMode": "cluster",
    "executorCores": 2,
    "executorMemory": "2g",
    "numExecutors": 2
  }
}

4.2.2 参数解析与处理

系统在operateInputParameter()方法中会自动补充必要的系统参数：

private void operateInputParameter(Map<String, String> inputParameter,
                                 DataQualityTaskExecutionContext dataQualityTaskExecutionContext) {
    // 添加规则元数据
    inputParameter.put(RULE_ID, String.valueOf(dataQualityTaskExecutionContext.getRuleId()));
    inputParameter.put(RULE_TYPE, String.valueOf(dataQualityTaskExecutionContext.getRuleType()));
    inputParameter.put(RULE_NAME, ArgsUtils.wrapperSingleQuotes(dataQualityTaskExecutionContext.getRuleName()));
    
    // 添加流程实例信息
    inputParameter.put(PROCESS_DEFINITION_ID, String.valueOf(dqTaskExecutionContext.getProcessDefineId()));
    inputParameter.put(PROCESS_INSTANCE_ID, String.valueOf(dqTaskExecutionContext.getProcessInstanceId()));
    inputParameter.put(TASK_INSTANCE_ID, String.valueOf(dqTaskExecutionContext.getTaskInstanceId()));
    
    // 设置错误数据输出路径
    inputParameter.put(ERROR_OUTPUT_PATH, 
        dataQualityTaskExecutionContext.getHdfsPath() + SLASH + 
        dqTaskExecutionContext.getProcessDefineId() + UNDERLINE +
        dqTaskExecutionContext.getProcessInstanceId() + UNDERLINE +
        dqTaskExecutionContext.getTaskName());
}

4.2.3 生成的Spark命令

最终构建的Spark提交命令如下：

${SPARK_HOME}/bin/spark-submit \
--class org.apache.dolphinscheduler.data.quality.DataQualityApplication \
--master yarn \
--deploy-mode cluster \
--executor-cores 2 \
--executor-memory 2g \
--num-executors 2 \
/path/to/dolphinscheduler-data-quality.jar \
"{\"ruleId\":4,\"source\":{\"connector\":\"mysql\",\"jdbcUrl\":\"jdbc:mysql://host:port/user_db\",\"table\":\"t_user\",\"username\":\"xxx\",\"password\":\"xxx\"},\"comparison\":{\"type\":\"REGEX\",\"column\":\"email\",\"regexp\":\"^[A-Za-z0-9+_.-]+@[A-Za-z0-9.-]+$\",\"threshold\":0}}"

4.3 表对比校验配置

表对比校验（ruleId=5）用于验证两个表的数据一致性，适用于数据同步后的校验场景：

{
  "ruleId": 5,
  "ruleInputParameter": {
    "sourceTable": "t_order",
    "targetTable": "t_order_backup",
    "keyColumn": "order_id",
    "compareColumns": "order_no,amount,status",
    "maxDiffCount": "10"
  },
  "sparkParameters": {
    "master": "yarn",
    "deployMode": "cluster",
    "executorCores": 4,
    "executorMemory": "4g",
    "numExecutors": 3
  }
}

5. 结果处理与告警机制

5.1 结果数据存储

数据质量校验结果包含两部分：

1. 校验统计信息：存储于元数据库的dq_result表
2. 错误数据明细：存储于HDFS路径，路径格式为：

   ${hdfsPath}/${processDefineId}_${processInstanceId}_${taskName}
   

5.2 告警通知

当校验结果超过设定阈值时，系统会自动触发告警。告警实现位于ProcessAlertManager.java：

public void sendDataQualityTaskExecuteResultAlert(DqExecuteResult result, ProcessInstance processInstance) {
    String state = result.getState() == DqState.SUCCESS ? "success" : "failed";
    AlertInfo alert = new AlertInfo();
    alert.setTitle("DataQualityResult [" + result.getTaskName() + "] " + state);
    alert.setContent(getDataQualityAlterContent(result));
    alert.setAlertGroupId(processInstance.getAlertGroupId());
    alert.setProcessDefineId(processInstance.getProcessDefineId());
    alert.setProcessInstanceId(processInstance.getId());
    alert.setTaskInstanceId(result.getTaskInstanceId());
    alert.setAlertType(AlertType.DATA_QUALITY);
    
    alertManager.sendAlert(alert);
}

告警内容格式如下：

数据质量校验结果通知:
任务名称: user_email_format_check
规则名称: 邮箱格式验证
执行时间: 2025-09-14 10:30:00
校验状态: 失败
失败原因: 发现23条记录不符合规则
影响行数: 23
错误详情: hdfs:///dolphinscheduler/dq/error/10086_202309141030_user_email_format_check

6. 性能优化策略

6.1 Spark参数调优

针对不同数据量和校验类型，需要合理调整Spark参数：

数据规模	executorCores	executorMemory	numExecutors	driverMemory
小表(<100万)	2	2g	2-3	1g
中表(100万-1亿)	4	4-8g	3-5	2g
大表(>1亿)	8	8-16g	5-10	4g

6.2 增量校验配置

通过DATA_TIME参数实现增量校验，仅检查指定时间范围内的数据：

{
  "ruleInputParameter": {
    "column": "create_time",
    "minValue": "${system.biz.date}",
    "maxValue": "${system.biz.date + 1}"
  }
}

6.3 分桶表优化

对于大表对比场景，建议使用分桶表并指定分桶列，提高对比效率：

{
  "ruleInputParameter": {
    "bucketColumn": "order_date",
    "bucketNum": "10"
  }
}

7. 高级应用：自定义验证规则

7.1 规则扩展机制

DolphinScheduler支持通过插件方式扩展自定义验证规则，需实现以下接口：

1. 创建规则定义类，继承AbstractRuleParameter
2. 实现规则执行逻辑，继承AbstractRuleExecutor
3. 在RuleManager中注册自定义规则

7.2 自定义规则开发步骤

timeline
    title 自定义规则开发流程
    2025-09-14 : 创建规则参数类(继承AbstractRuleParameter)
    2025-09-14 : 实现规则执行器(继承AbstractRuleExecutor)
    2025-09-15 : 配置规则元数据(rule-configuration.json)
    2025-09-15 : 打包为插件JAR
    2025-09-16 : 部署插件并测试

8. 最佳实践与注意事项

8.1 任务编排建议

1. 前置依赖：数据质量任务应依赖于数据加载任务，确保数据已准备就绪
2. 并行执行：不同表的校验任务可并行执行，提高整体效率
3. 失败处理：关键校验任务失败后应终止后续流程，避免脏数据流转

flowchart LR
    A[数据抽取] --> B[数据清洗]
    B --> C[数据加载]
    C --> D{并行校验任务}
    D --> E[用户表校验]
    D --> F[订单表校验]
    D --> G[产品表校验]
    E & F & G --> H{全部成功?}
    H -->|是| I[数据汇总]
    H -->|否| J[触发告警与回滚]

8.2 常见问题与解决方案

问题	原因	解决方案
Spark任务提交失败	Spark_HOME配置错误	检查worker节点的Spark环境变量
数据源连接失败	数据库驱动缺失	在lib目录添加对应数据库驱动JAR
正则表达式不生效	特殊字符未转义	使用StringEscapeUtils.escapeJava()处理正则表达式
大表校验性能差	未设置分区列	配置partitionColumn和numPartitions参数
结果数据过大	错误数据太多	增加过滤条件，只保留关键错误字段

9. 总结与展望

DolphinScheduler的数据质量校验任务为数据处理流程提供了强大的质量保障机制，通过内置的多种验证器和灵活的配置方式，帮助用户轻松实现数据质量监控。随着数据量的爆炸式增长，数据质量将成为越来越重要的议题。未来版本将进一步增强：

1. 实时数据质量监控能力
2. AI辅助的异常检测与修复建议
3. 更丰富的可视化报表
4. 与数据血缘的深度集成

通过本文的指南，您应该已经掌握了DolphinScheduler数据质量校验任务的核心使用方法。建议从关键业务表的基础校验规则开始实践，逐步构建完整的数据质量监控体系。

附录：快速入门命令

克隆项目仓库

git clone https://gitcode.com/gh_mirrors/do/dolphinscheduler.git
cd dolphinscheduler

构建数据质量模块

mvn clean package -pl dolphinscheduler-task-plugin/dolphinscheduler-task-dataquality -am -DskipTests

本地运行示例任务

./bin/dolphinscheduler-cli task execute -t DATA_QUALITY -p "{\"ruleId\":1,\"ruleInputParameter\":{\"column\":\"id\",\"threshold\":\"100\"},\"sparkParameters\":{\"master\":\"local[2]\"}}"