Apache Dolphinscheduler数据质量源码分析

Apache DolphinScheduler 是一个分布式、易扩展的可视化数据工作流使命调度系统，广泛应用于数据调度和处置惩罚领域。
在大规模数据工程项目中，数据质量的管理至关紧张，而 DolphinScheduler 也提供了数据质量检查的计算能力。本文将对 Apache DolphinScheduler 的数据质量模块举行源码分析，资助开发者深入明白其背后的实现原理与设计理念。
数据质量规则

Apache Dolphinscheduler 数据质量模块支持多种常用的数据质量规则，如下图所示。

数据质量规则主要包罗空值校验、自定义SQL、跨表准确性、跨表值比、字段长度校验、唯一性校验、及时性检查、枚举值校验、表行数校验等。
数据质量工作流程

数据质量运行流程分为2个部分：
（1）在Web端举行数据质量检测的流程定义，通过DolphinScheduer举行调度，提交到Spark计算引擎；
（2）Spark端负责解析数据质量模型的参数，通过读取数据、实行转换、输出三个步骤，完成数据质量检测使命，工作流程如下图所示。

在Web端举行定义

数据质量定义如下图所示，这里只定义了一个节点。

以一个空值检测的输入参数为例，在界面完成配置后，会生产一个JSON文件。
这个JSON文件会以字符串参数形式提交给Spark集群，举行调度和计算。
JSON文件如下所示。

1. {
2.     "name": "$t(null_check)",
3.     "env": {
4.         "type": "batch",
5.         "config": null
6.     },
7.     "readers": [
8.         {
9.             "type": "JDBC",
10.             "config": {
11.                 "database": "ops",
12.                 "password": "***",
13.                 "driver": "com.mysql.cj.jdbc.Driver",
14.                 "user": "root",
15.                 "output_table": "ops_ms_alarm",
16.                 "table": "ms_alarm",
17.                 "url": "jdbc:mysql://192.168.3.211:3306/ops?allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false"
18.             }
19.         }
20.     ],
21.     "transformers": [
22.         {
23.             "type": "sql",
24.             "config": {
25.                 "index": 1,
26.                 "output_table": "total_count",
27.                 "sql": "SELECT COUNT(*) AS total FROM ops_ms_alarm"
28.             }
29.         },
30.         {
31.             "type": "sql",
32.             "config": {
33.                 "index": 2,
34.                 "output_table": "null_items",
35.                 "sql": "SELECT * FROM ops_ms_alarm WHERE (alarm_time is null or alarm_time = '') "
36.             }
37.         },
38.         {
39.             "type": "sql",
40.             "config": {
41.                 "index": 3,
42.                 "output_table": "null_count",
43.                 "sql": "SELECT COUNT(*) AS nulls FROM null_items"
44.             }
45.         }
46.     ],
47.     "writers": [
48.         {
49.             "type": "JDBC",
50.             "config": {
51.                 "database": "dolphinscheduler3",
52.                 "password": "***",
53.                 "driver": "com.mysql.cj.jdbc.Driver",
54.                 "user": "root",
55.                 "table": "t_ds_dq_execute_result",
56.                 "url": "jdbc:mysql://192.168.3.212:3306/dolphinscheduler3?characterEncoding=utf-8&allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false",
57.                 "sql": "select 0 as rule_type,'$t(null_check)' as rule_name,0 as process_definition_id,25 as process_instance_id,26 as task_instance_id,null_count.nulls AS statistics_value,total_count.total AS comparison_value,7 AS comparison_type,3 as check_type,0.95 as threshold,3 as operator,1 as failure_strategy,'hdfs://xmaster:9000/user/hadoop/data_quality_error_data/0_25_211-ops-ms_alarm-空值检测' as error_output_path,'2022-11-16 03:40:32' as create_time,'2022-11-16 03:40:32' as update_time from null_count full join total_count"
58.             }
59.         },
60.         {
61.             "type": "JDBC",
62.             "config": {
63.                 "database": "dolphinscheduler3",
64.                 "password": "***",
65.                 "driver": "com.mysql.cj.jdbc.Driver",
66.                 "user": "root",
67.                 "table": "t_ds_dq_task_statistics_value",
68.                 "url": "jdbc:mysql://192.168.3.212:3306/dolphinscheduler3?characterEncoding=utf-8&allowLoadLocalInfile=false&autoDeserialize=false&allowLocalInfile=false&allowUrlInLocalInfile=false",
69.                 "sql": "select 0 as process_definition_id,26 as task_instance_id,1 as rule_id,'ZKTZKDBTRFDKXKQUDNZJVKNX8OIAEVLQ91VT2EXZD3U=' as unique_code,'null_count.nulls'AS statistics_name,null_count.nulls AS statistics_value,'2022-11-16 03:40:32' as data_time,'2022-11-16 03:40:32' as create_time,'2022-11-16 03:40:32' as update_time from null_count"
70.             }
71.         },
72.         {
73.             "type": "hdfs_file",
74.             "config": {
75.                 "path": "hdfs://xmaster:9000/user/hadoop/data_quality_error_data/0_25_211-ops-ms_alarm-空值检测",
76.                 "input_table": "null_items"
77.             }
78.         }
79.     ]
80. }

复制代码

Spark端源码分析

DataQualityApplication.java 是Spark程序入口

1. public static void main(String[] args) throws Exception {
2.     //...
3. //从命令行获取参数
4.     String dataQualityParameter = args[0];
5. //   将json参数转为DataQualityConfiguration对象
6.     DataQualityConfiguration dataQualityConfiguration = JsonUtils.fromJson(dataQualityParameter,DataQualityConfiguration.class);
7.     //...
8. //构建 SparkRuntimeEnvironment的参数Config对象
9.     EnvConfig envConfig = dataQualityConfiguration.getEnvConfig();
10.     Config config = new Config(envConfig.getConfig());
11.     config.put("type",envConfig.getType());
12.     if (Strings.isNullOrEmpty(config.getString(SPARK_APP_NAME))) {
13.         config.put(SPARK_APP_NAME,dataQualityConfiguration.getName());
14.     }
16.     SparkRuntimeEnvironment sparkRuntimeEnvironment = new SparkRuntimeEnvironment(config);
17. //委托给 DataQualityContext执行
18.     DataQualityContext dataQualityContext = new DataQualityContext(sparkRuntimeEnvironment,dataQualityConfiguration);
19.     dataQualityContext.execute();
20. }

复制代码

数据质量配置类

1. public class DataQualityConfiguration implements IConfig {
2.     @JsonProperty("name")
3.     private String name; // 名称
4.     @JsonProperty("env")
5.     private EnvConfig envConfig; // 环境配置
6.     @JsonProperty("readers")
7.     private List<ReaderConfig> readerConfigs; // reader配置
8.     @JsonProperty("transformers")
9.     private List<TransformerConfig> transformerConfigs;  // transformer配置
10.     @JsonProperty("writers")
11.     private List<WriterConfig> writerConfigs; // writer配置
12. //...
13. }

复制代码

DataQualityContext#execute从dataQualityConfiguration类中获取Readers、Transformers、Writers， 委托给SparkBatchExecution实行

1. public void execute() throws DataQualityException {
2. // 将List<ReaderConfig>转为List<BatchReader>
3.     List<BatchReader> readers = ReaderFactory
4.             .getInstance()
5.             .getReaders(this.sparkRuntimeEnvironment,dataQualityConfiguration.getReaderConfigs());
6. // 将List<TransformerConfig>转为List<BatchTransformer>
7.     List<BatchTransformer> transformers = TransformerFactory
8.             .getInstance()
9.             .getTransformer(this.sparkRuntimeEnvironment,dataQualityConfiguration.getTransformerConfigs());
10. // 将List<WriterConfig>转为List<BatchWriter>
11.     List<BatchWriter> writers = WriterFactory
12.             .getInstance()
13.             .getWriters(this.sparkRuntimeEnvironment,dataQualityConfiguration.getWriterConfigs());
14. // spark 运行环境
15.     if (sparkRuntimeEnvironment.isBatch()) {
16. // 批模式
17.         sparkRuntimeEnvironment.getBatchExecution().execute(readers,transformers,writers);
18.     } else {
19. // 流模式， 暂不支持
20.         throw new DataQualityException("stream mode is not supported now");
21.     }
22. }

复制代码

目前 Apache DolphinScheduler 暂时不支持及时数据的质量检测。
ReaderFactory类采用了单例和工厂方法的设计模式，目前支持JDBC和HIVE的数据源的读取， 对应Reader类HiveReader、JDBCReader。
WriterFactory类采用了单例和工厂方法的设计模式，目前支持JDBC、HDFS、LOCAL_FILE的数据源的输出，对应Writer类JdbcWriter、 HdfsFileWriter和 LocalFileWriter 。
TransformerFactory类采用了单例和工厂方法的设计模式，目前仅支持TransformerType.SQL的转换器类型。
结合JSON可以看出一个空值检测的Reader、Tranformer、 Writer环境:

1个Reader ：读取源表数据
3个Tranformer：
total_count 行总数
null_items 空值项（行数据）
null_count （空值数）

计算SQL如下

1. -- SELECT COUNT(*) AS total FROM ops_ms_alarm
2. -- SELECT * FROM ops_ms_alarm WHERE (alarm_time is null or alarm_time = '')
3. -- SELECT COUNT(*) AS nulls FROM null_items

复制代码

3个Writer：第一个是JDBC Writer， 将比较值、统计值输出t\_ds\_dq\_execute\_result 数据质量实行结果表。

1. SELECT
2.    //...
3.     null_count.nulls AS statistics_value,
4.     total_count.total AS comparison_value,
5.    //...
6.     'hdfs://xmaster:9000/user/hadoop/data_quality_error_data/0_25_211-ops-ms_alarm-空值检测' AS error_output_path,
7.    //...
8. FROM
9.     null_count
10.     FULL JOIN total_count

复制代码

第二个是JDBC Writer，将statistics\_value写入到表 t\_ds\_dq\_task\_statistics\_value

1. SELECT
2.     //...
3.     //...
4.     'null_count.nulls' AS statistics_name,
5.     null_count.nulls AS statistics_value,
6.     //...
7. FROM
8.     null_count

复制代码

第3个是HDFS Writer，将空值项写入到HDFS文件目录

1. {
2.     "type": "hdfs_file",
3.     "config": {
4.         "path": "hdfs://xmaster:9000/user/hadoop/data_quality_error_data/0_25_211-ops-ms_alarm-空值检测",
5.         "input_table": "null_items"
6.     }
7. }

复制代码

SparkBatchExecution#execute

1. public class SparkBatchExecution implements Execution<BatchReader, BatchTransformer, BatchWriter> {
2.     private final SparkRuntimeEnvironment environment;
3.     public SparkBatchExecution(SparkRuntimeEnvironment environment) throws ConfigRuntimeException {
4.         this.environment = environment;
5.     }
6.    
7.     @Override
8.     public void execute(List<BatchReader> readers, List<BatchTransformer> transformers, List<BatchWriter> writers) {
9. // 为每一个reader注册输入临时表
10.         readers.forEach(reader -> registerInputTempView(reader, environment));
12.         if (!readers.isEmpty()) {
13. // 取readers列表的第一个reader读取数据集合， reader的实现类有HiveReader、JdbcReader
14.             Dataset<Row> ds = readers.get(0).read(environment);
15.             for (BatchTransformer tf:transformers) {
16. // 执行转换
17.                 ds = executeTransformer(environment, tf, ds);
18. // 将转换后结果写到临时表
19.                 registerTransformTempView(tf, ds);
20.             }
22.             for (BatchWriter sink: writers) {
23. // 执行将转换结果由writer输出， writer的实现类有JdbcWriter、LocalFileWriter、HdfsFileWriter
24.                 executeWriter(environment, sink, ds);
25.             }
26.         }
27. // 结束
28.         environment.sparkSession().stop();
29.     }
30. }

复制代码

SparkBatchExecution#registerInputTempView

1. //注册输入临时表, 临时表表名为OUTPUT_TABLE的名字
2.     private void registerInputTempView(BatchReader reader, SparkRuntimeEnvironment environment) {
3.         Config conf = reader.getConfig();
4.         if (Boolean.TRUE.equals(conf.has(OUTPUT_TABLE))) {// ops_ms_alarm
5.             String tableName = conf.getString(OUTPUT_TABLE);        
6.             registerTempView(tableName, reader.read(environment));
7.         } else {
8.             throw new ConfigRuntimeException(
9.                 "[" + reader.getClass().getName() + "] must be registered as dataset, please set "output_table" config");
10.         }
11.     }

复制代码

调用Dataset.createOrReplaceTempView方法

1. private void registerTempView(String tableName, Dataset<Row> ds) {
2.     if (ds != null) {
3.         ds.createOrReplaceTempView(tableName);
4.     } else {
5.         throw new ConfigRuntimeException("dataset is null, can not createOrReplaceTempView");
6.     }
7. }

复制代码

实行转换executeTransformer

1. private Dataset<Row> executeTransformer(SparkRuntimeEnvironment environment, BatchTransformer transformer, Dataset<Row> dataset) {
2.     Config config = transformer.getConfig();
3.     Dataset<Row> inputDataset;
4.     Dataset<Row> outputDataset = null;
5.     if (Boolean.TRUE.equals(config.has(INPUT_TABLE))) {
6. // 从INPUT_TABLE获取表名
7.         String[] tableNames = config.getString(INPUT_TABLE).split(",");
9. // outputDataset合并了inputDataset数据集合
10.         for (String sourceTableName: tableNames) {
11.             inputDataset = environment.sparkSession().read().table(sourceTableName);
12.             if (outputDataset == null) {
13.                 outputDataset = inputDataset;
14.             } else {
15.                 outputDataset = outputDataset.union(inputDataset);
16.             }
17.         }
18.     } else {
19. //  配置文件无INPUT_TABLE
20.         outputDataset = dataset;
21.     }
22. // 如果配置文件中配置了TMP_TABLE， 将outputDataset 注册到TempView
23.     if (Boolean.TRUE.equals(config.has(TMP_TABLE))) {
24.         if (outputDataset == null) {
25.             outputDataset = dataset;
26.         }
27.         String tableName = config.getString(TMP_TABLE);
28.         registerTempView(tableName, outputDataset);
29.     }
30. //  转换器进行转换
31.     return transformer.transform(outputDataset, environment);
32. }

复制代码

SqlTransformer#transform 终极是使用spark-sql举行处置惩罚， 所以焦点照旧这个SQL语句，SQL需要在web端生成好，参考前面的JSON文件。

1. public class SqlTransformer implements BatchTransformer {
2.     private final Config config;
3.     public SqlTransformer(Config config) {
4.         this.config = config;
5.     }
6. //...
7.     @Override
8.     public Dataset<Row> transform(Dataset<Row> data, SparkRuntimeEnvironment env) {
9.         return env.sparkSession().sql(config.getString(SQL));
10.     }
11. }

复制代码

将数据输出到指定的位置executeWriter

1. private void executeWriter(SparkRuntimeEnvironment environment, BatchWriter writer, Dataset<Row> ds) {
2.     Config config = writer.getConfig();
3.     Dataset<Row> inputDataSet = ds;
4.     if (Boolean.TRUE.equals(config.has(INPUT_TABLE))) {
5.         String sourceTableName = config.getString(INPUT_TABLE);
6.         inputDataSet = environment.sparkSession().read().table(sourceTableName);
7.     }
9.     writer.write(inputDataSet, environment);
10. }

复制代码

总体来讲，Apache Dolphinscheduler的数据质量检测实现相对简朴明了，只要采用Spark SQL举行计算。在本文中，我们深入分析了数据质量模块的源码布局和实现逻辑，Apache DolphinScheduler 数据质量模块的设计理念强调灵活性和扩展性，这使得它可以适应不同企业的多样化需求。
对于开发者而言，深入明白其源码不仅有助于更好地使用 DolphinScheduler，也为进一步扩展其功能提供了方向和灵感。希望本文能够为您在数据质量控制和开源项目深入探索方面提供资助。
​

本文由 白鲸开源 提供发布支持！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。