手写一个DQC(一)DQC简介及数据解析


一:DQC核心流程

    Define:数据质检规则(指标)的定义。
            你要告警给谁,你要使用什么方式告警(邮件,即时消息),你的规则是什么(空值,波动)等
    Measure:数据质检任务的执行
            数据在哪存储:hive、mysql是基本的数据库、CK、kylin等
    Analyze:数据质检结果量化及可视化展示。
            分为两种情况:(一)不需要图形化界面,直接在调度里面进行bash配置,使用自定义代码解析(二)有图形化界面,可以操作和查看历史结果

二:DQC标准

    Accuracy:准确性。如是否符合表的加工逻辑。
Completeness:完备性。如数据是否存在丢失。
Timeliness:及时性。如表数据是否按时产生。
Uniqueness:唯一性。如主键字段是否唯一。
Validity:合规性。如字段长度是否合规、枚举值集合是否合规。
Consistency:一致性。如表与表之间在某些字段上是否存在矛盾。

三:DQC规则

2.1有效性
字段长度有效、字段内容有效、字段数值范围有效、枚举值个数有效、枚举值集合有效
2.2 唯一性
对主键是否存在重复数据的监控指标。
2.3 完整性
字段是否为空或NULL、记录数是否丢失、记录数环比波动、录数波动范围、记录数方差检验、
2.4 准确性
数值同比、数值环比、数值方差检验、表逻辑检查
2.5 一致性
表级别一致性检查,外键检查
2.6 时效性
表级别质量监控指标,数据是否按时产出
2.7数据剖析
最大值检查、最小值检查、平均值检查、汇总值检查
2.8 自定义规则检查
用户写自定义SQL实现的监控规则
从有效性、唯一性、完整性、准确性、一致性、时效性、数据剖析和自定义规则检查等几个维度对数据质量进行测量,但对于现在超级大的数据量级监控所有的数据是不符合成本效率的。
因此,知道哪些数据为最关键的,对这些关键数据进行全链路的数据质量,这样有助于防止错误或揭示改进的机会。

总结:指定值、空值、外键规范、外键最大最小、行数统计、最大、最小、平均、用户自定义

四:样例代码

           

String[] argsa={"-alter_user=xiaolong.wu","-alter_type=3","-counttype=A0001","-count_column=tag","-databases=hive.tranadm.adm_fin_paytrigger_revenue_ds","-dt=20220509","-filter_column=easysolar|tag","-filter_column_value=1000|上一节点","-assert_type=eq","-assert_Stringue=200","-assert_rate=0.2"};

MyArgs myArgs = MyArgs.build(argsa);

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

class MyArgsItemInfo{
    private String optionName  ="";
    private boolean necessary  = false;
    private String datatype  = "";
    private String dataverfiy  = ""; //  String:正则表达式   enum:|分割的枚举类型    float:最大值|最小值   date:日期格式
    private String desc  ="";

    public MyArgsItemInfo(String optionName, boolean necessary,String datetype,String dateverfiy, String desc) {
        this.optionName = optionName;
        this.necessary = necessary;
        this.datatype = datetype;
        this.dataverfiy = dateverfiy;
        this.desc = desc;
    }
    public String getOptionName() {
        return optionName;
    }
    public String getDatatype() {
        return datatype;
    }
    public String getDataverfiy() {
        return dataverfiy;
    }
}

class MyArgsInfo {

    public static MyArgsInfoBuildFactory init(){
        return new MyArgsInfoBuildFactory();
    }
    public static class MyArgsInfoBuildFactory {

        private List buildFactoryerMyArgsItemInfo = new ArrayList();

        public List build(){
            return buildFactoryerMyArgsItemInfo;
        }
        public MyArgsInfoBuildFactory addMyArgsItemInfo(String optionName, boolean necessary,String datetype,String dateverfiy, String desc) {
            buildFactoryerMyArgsItemInfo.add(new MyArgsItemInfo(optionName,necessary,datetype,dateverfiy,desc));
            return this;
        }
    }
}

public class MyArgs{
    private String alter_user = "";              //告警接收人
    private String alter_type = "";              //邮件,企业微信
    private String counttype = "";               //A0003
    private String count_column = "";            //
    private String databases  = "";              // hive.ods.ods_user_active_di  限定数据库类型、库名、表明
    private String dt = "";
    private String filter_column = "";           //,分割    接受多个字段
    private String filter_column_value = "";   //,分割    接受多个字段
    private String sql = "";
    private String assert_type = "";             //eq(==), lt(<), gt(>), le(<=), ge(>=), ne(!=)
    private String assert_Stringue = "";           //,
    private String assert_rate = "";               //,

    public static MyArgs build(String[] args) {
        return new MyArgsBuildFactory().build(args);
    }

    public MyArgs(MyArgsBuildFactory builder) {
        this.alter_user = builder.alter_user;
        this.alter_type = builder.alter_type ;
        this.counttype = builder.counttype;
        this.count_column = builder.count_column;
        this.databases = builder.databases;
        this.dt = builder.dt;
        this.filter_column = builder.filter_column;
        this.filter_column_value = builder.filter_column_value;
        this.sql = builder.sql;
        this.assert_type = builder.assert_type;
        this.assert_Stringue = builder.assert_Stringue;
        this.assert_rate = builder.assert_rate;
    }

    public static class MyArgsBuildFactory{
        private boolean checkFalse = false;

        private String alter_user = "";              //告警接收人
        private String alter_type = "";              //邮件,企业微信
        private String counttype = "";               //count(1) count(distinct )
        private String count_column = "";            //
        private String databases  = "";              // hive.ods.ods_user_active_di  限定数据库类型、库名、表明
        private String dt = "";                      // 默认比较今天和前一天的结果
        private String filter_column = "";           //,分割    接受多个字段
        private String filter_column_value = "";   //,分割    接受多个字段
        private String sql = "";
        private String assert_type = "";             //eq(==), lt(<), gt(>), le(<=), ge(>=), ne(!=)
        private String assert_Stringue = "";           //,
        private String assert_rate = "";               //,

        public void checkArgs(List myArgsInfo){
            for (int i = 0; i < myArgsInfo.size(); i++) {
                MyArgsItemInfo myArgsItemInfo = myArgsInfo.get(i);
                String datatype = myArgsItemInfo.getDatatype();
                String datavalue = getValueByName(myArgsItemInfo.getOptionName());
                //  String:正则表达式   enum:|分割的枚举类型    float:最大值|最小值   date:日期格式
                if(!datavalue.equals("")){   //空值的去掉
                    switch (datatype){
                        case "String" : {
                            if(!datavalue.matches(myArgsItemInfo.getDataverfiy())){
                                System.out.println(myArgsItemInfo.getOptionName()+"参数值存在问题,非"+myArgsItemInfo.getDataverfiy()+"正则表达式");
                                checkFalse = true;
                            }
                        };break;
                        case "enum" : {
                            if(!Arrays.asList(myArgsItemInfo.getDataverfiy().split("\\|")).contains(datavalue)){
                                System.out.println(myArgsItemInfo.getOptionName()+"参数值存在问题,非"+myArgsItemInfo.getDataverfiy()+"枚举值");
                                checkFalse = true;
                            }
                        }break;
                        case "float" : {
                            float minvalue = Float.valueOf(myArgsItemInfo.getDataverfiy().split("\\|")[0]);
                            float maxvalue = Float.valueOf(myArgsItemInfo.getDataverfiy().split("\\|")[0]);
                            if(Float.valueOf(datavalue)>maxvalue || Float.valueOf(datavalue) myArgsItemInfoInit(){
            MyArgsInfo.MyArgsInfoBuildFactory myArgsInfoBuildFactory = MyArgsInfo.init();
            myArgsInfoBuildFactory.addMyArgsItemInfo("alter_user", true,"String","[a-z]*.[a-z]*", "告警接收人")
                    .addMyArgsItemInfo("alter_type", true,"enum","1|2|3", "1:邮件,2:企业微信,3邮件+企业微信")
                    .addMyArgsItemInfo("counttype", true,"enum","A0001|A0002|A0003|A0004|A0005", "计算类型")
                    .addMyArgsItemInfo("count_column", true,"String","[a-zA-Z0-9_]*", "列名")
                    .addMyArgsItemInfo("databases", true,"String","(hive|mysql|kylin|clickhouse|hbase).[a-z_]*.[a-z_]*", "hive.ods.ods_user_active_di  限定数据库类型、库名、表明")
                    .addMyArgsItemInfo("dt", false,"date","yyyyMMdd", "时间")
                    .addMyArgsItemInfo("filter_column", true,"String","[a-zA-Z0-9_\\|]*", "|分割    接受多个字段")
                    .addMyArgsItemInfo("filter_column_value", true,"String","[a-zA-Z0-9_\\|]*", "|分割    接受多个字段")
                    .addMyArgsItemInfo("sql", false,"String","[a-zA-Z\\*\\s]*", "自定义SQL")
                    .addMyArgsItemInfo("assert_type", true,"enum","eq|lt|gt|le|ge|ne", "eq(==), lt(<), gt(>), le(<=), ge(>=), ne(!=)")
                    .addMyArgsItemInfo("assert_Stringue", true,"float","-2099999999|2099999999", "限定数值")
                    .addMyArgsItemInfo("assert_rate", true,"float","0|1", "限定比率");
            return myArgsInfoBuildFactory.build();
        }
        public void myArgsValueInit(String[] args){
            for (int i = 0; i < args.length; i++) {
                String argString = args[i].substring(1,args[i].length());
                String argName = argString.split("=")[0];
                String argValue = argString.split("=")[1];

                switch (argName){
                    case "alter_user" : this.alter_user(argValue);break;
                    case "alter_type" : this.alter_type(argValue);break;
                    case "counttype" : this.counttype(argValue);break;
                    case "count_column" : this.count_column(argValue);break;
                    case "databases" : this.databases(argValue);break;
                    case "dt" : this.dt(argValue);break;
                    case "filter_column" : this.filter_column(argValue);break;
                    case "filter_column_value" : this.filter_column_value(argValue);break;
                    case "sql" : this.sql(argValue);break;
                    case "assert_type" : this.assert_type(argValue);break;
                    case "assert_Stringue" : this.assert_Stringue(argValue);break;
                    case "assert_rate" : this.assert_rate(argValue);break;
                }
            }
        }
        //主要方法
        public MyArgs build(String[] args){
            List myArgsInfo = myArgsItemInfoInit();  //将列信息初始化
            myArgsValueInit(args);                                   //将参数命令行过来的值初始化
            checkArgs(myArgsInfo);                                   //检查命令行过来的值是否符合  列表达式
            return new MyArgs(this);
        }


        public MyArgsBuildFactory alter_user(String alter_user) {
            this.alter_user = alter_user;
            return this;
        }
        public MyArgsBuildFactory alter_type(String alter_type) {
            this.alter_type = alter_type;
            return this;
        }
        public MyArgsBuildFactory counttype(String counttype) {
            this.counttype = counttype;
            return this;
        }
        public MyArgsBuildFactory count_column(String count_column) {
            this.count_column = count_column;
            return this;
        }
        public MyArgsBuildFactory databases(String databases) {
            this.databases = databases;
            return this;
        }
        public MyArgsBuildFactory dt(String dt) {
            this.dt = dt;
            return this;
        }
        public MyArgsBuildFactory filter_column(String filter_column) {
            this.filter_column = filter_column;
            return this;
        }
        public MyArgsBuildFactory filter_column_value(String filter_column_value) {
            this.filter_column_value = filter_column_value;
            return this;
        }
        public MyArgsBuildFactory sql(String sql) {
            this.sql = sql;
            return this;
        }
        public MyArgsBuildFactory assert_type(String assert_type) {
            this.assert_type = assert_type;
            return this;
        }
        public MyArgsBuildFactory assert_Stringue(String assert_Stringue) {
            this.assert_Stringue = assert_Stringue;
            return this;
        }
        public MyArgsBuildFactory assert_rate(String assert_rate) {
            this.assert_rate = assert_rate;
            return this;
        }
    }
    public String getAlter_user() {
        return alter_user;
    }
    public String getAlter_type() {
        return alter_type;
    }
    public String getCounttype() {
        return counttype;
    }
    public String getCount_column() {
        return count_column;
    }
    public String getDatabases() {
        return databases;
    }
    public String getDt() {
        return dt;
    }
    public String getFilter_column() {
        return filter_column;
    }
    public String getFilter_column_value() {
        return filter_column_value;
    }
    public String getSql() {
        return sql;
    }
    public String getAssert_type() {
        return assert_type;
    }
    public String getAssert_Stringue() {
        return assert_Stringue;
    }
    public String getAssert_rate() {
        return assert_rate;
    }
    @Override
    public String toString() {
        return "MyArgs{" +
                "alter_user='" + alter_user + '\'' +
                ", alter_type=" + alter_type +
                ", counttype='" + counttype + '\'' +
                ", count_column='" + count_column + '\'' +
                ", databases='" + databases + '\'' +
                ", dt='" + dt + '\'' +
                ", filter_column='" + filter_column + '\'' +
                ", filter_column_value='" + filter_column_value + '\'' +
                ", sql='" + sql + '\'' +
                ", assert_type='" + assert_type + '\'' +
                ", assert_Stringue=" + assert_Stringue +
                ", assert_rate=" + assert_rate +
                '}';
    }
}