数据质量概述及通用脚本增量数据质量检验通用脚本（指定分区数据质量检验）全量数据质量检验通用脚本ODS层数据质量校验DWD层数据质量校验

74 阅读 0 评论 49 点赞

我是靠谱客的博主沉静蚂蚁，这篇文章主要介绍数据质量概述及通用脚本增量数据质量检验通用脚本（指定分区数据质量检验）全量数据质量检验通用脚本ODS层数据质量校验DWD层数据质量校验，现在分享给大家，希望可以做个参考。

概述

数据质量表示着数据的健康性。

数据质量的高低代表了该数据满足数据消费者期望的程度，这种程度基于他们对数据的使用预期，也就是数据的准确程度是否达到他们的要求。

而数据质量管理就是负责这个的模块。

数据质量标准分类

数据完整性
不存在大量或关键字段缺失值，在ETL过程中保证数据完整不丢失。数据总数增长符合正常趋势。
数据一致性
数仓各层数据应与上一层数据一致，数仓指标与数据源保持一致。
数据唯一性
每一个事实应当只出现一次。

增量数据质量检验通用脚本（指定分区数据质量检验）

表格式：

复制代码

1
2
3
4
5
6
7
8
9
10
create table table_increment(
    data_date string comment '数据时间分区dt',
    database_name string comment '库名',
    table_name string comment '表名',
    table_type string comment '表类型（全量/增量）',
    null_count bigint comment '表空值记录数',
    duplicate_count bigint  comment '表重复值记录数',
    add_count bigint comment '当日新增数量的记录数'
)

脚本：

复制代码

#!/bin/bash
# 校验数据的库名
database_name=$1
# 校验数据的表名
table_name=$2
# 需要校验空值的列名，以 ‘,’分割
null_column=$3
# 新增数据所在分区
do_date=$4

# 初始化SQL查询语句
null_where_sql_str=''
# 将空值校验字段切分为数组
array=(${null_column//,/ })
# 遍历数组，拼接空值查询条件
for(( i=0;i< ${#array[@]};i++)) do
    if [ $i -eq 0 ];then
        null_where_sql_str="where ${array[i]} is null"
    else
        null_where_sql_str="$null_where_sql_str or ${array[i]} is null"
    fi
done;

# 执行当日增量数据记录数量SQL查询语句
echo "----------检验当日增量记录数----------"
add_count_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select count(*) from $database_name.$table_name where start_date='$do_date'"`
# 取出当日增量数据记录数量
add_count_array=(${add_count_query_result//|/})
add_count=${add_count_array[3]}

# 执行当日增量空值数量记录数量SQL查询语句
echo "----------检验当日增量空值记录数----------"
total_null_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select count(*) from (select * from $database_name.$table_name where start_date='$do_date')increment $null_where_sql_str"`
# 取出当日增量空值数据记录数量
null_count_array=(${total_null_query_result//|/})
null_count=${null_count_array[3]}

# 执行当日增量重复值数量记录数量SQL查询语句
echo "----------检验当日增量重复值记录数----------"
table_duplicate_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select sum(tmp.duplicate_count) as duplicate_sum from(select count(*) as duplicate_count from (select * from $database_name.$table_name where start_date='$do_date')increment group by $null_column having count(*)>1) as tmp"`
# 取出当日增量数据记录数量
duplicate_count_array=(${table_duplicate_query_result//|/})
duplicate_count=${duplicate_count_array[3]}

# 将所有数据校验结果插入到表中
echo "----------开始插入数据----------"
beeline -u jdbc:hive2://node01:10000 -n hdfs -e "insert into datacheck.table_increment values('$do_date','$database_name','$table_name','increment_table',$null_count,$duplicate_count,$add_count)"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
#!/bin/bash
# 校验数据的库名
database_name=$1
# 校验数据的表名
table_name=$2
# 需要校验空值的列名，以 ‘,’分割
null_column=$3
# 新增数据所在分区
do_date=$4

# 初始化SQL查询语句
null_where_sql_str=''
# 将空值校验字段切分为数组
array=(${null_column//,/ })
# 遍历数组，拼接空值查询条件
for(( i=0;i< ${#array[@]};i++)) do
    if [ $i -eq 0 ];then
        null_where_sql_str="where ${array[i]} is null"
    else
        null_where_sql_str="$null_where_sql_str or ${array[i]} is null"
    fi
done;

# 执行当日增量数据记录数量SQL查询语句
echo "----------检验当日增量记录数----------"
add_count_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select count(*) from $database_name.$table_name where start_date='$do_date'"`
# 取出当日增量数据记录数量
add_count_array=(${add_count_query_result//|/})
add_count=${add_count_array[3]}

# 执行当日增量空值数量记录数量SQL查询语句
echo "----------检验当日增量空值记录数----------"
total_null_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select count(*) from (select * from $database_name.$table_name where start_date='$do_date')increment $null_where_sql_str"`
# 取出当日增量空值数据记录数量
null_count_array=(${total_null_query_result//|/})
null_count=${null_count_array[3]}

# 执行当日增量重复值数量记录数量SQL查询语句
echo "----------检验当日增量重复值记录数----------"
table_duplicate_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select sum(tmp.duplicate_count) as duplicate_sum from(select count(*) as duplicate_count from (select * from $database_name.$table_name where start_date='$do_date')increment group by $null_column having count(*)>1) as tmp"`
# 取出当日增量数据记录数量
duplicate_count_array=(${table_duplicate_query_result//|/})
duplicate_count=${duplicate_count_array[3]}

# 将所有数据校验结果插入到表中
echo "----------开始插入数据----------"
beeline -u jdbc:hive2://node01:10000 -n hdfs -e "insert into datacheck.table_increment values('$do_date','$database_name','$table_name','increment_table',$null_count,$duplicate_count,$add_count)"

四个参数：

第一个参数是数据库名
第二个参数是表名
第三个参数是字段名，字段名以逗号分隔
第四个参数是分区名

全量数据质量检验通用脚本

表格式：

复制代码

1
2
3
4
5
6
7
8
9
10
create table table_total_quantity(
    data_date string comment '数据时间分区dt',
    database_name string comment '库名',
    table_name string comment '表名',
    table_type string comment '表类型（全量/增量）',
    null_count bigint comment '表空值记录数',
    duplicate_count bigint  comment '表重复值记录数',
    total_count bigint comment '全表记录数'
)

脚本：

复制代码

#!/bin/bash
# 校验数据的库名
database_name=$1
# 校验数据的表名
table_name=$2
# 需要校验空值的列名，以 ‘,’分割
null_column=$3
# 日期
do_date=$4

# 执行当日全表数量记录数量SQL查询语句
echo "----------检验当日全量记录数----------"
total_count_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select count(*) from $database_name.$table_name"`
# 取出当日全量数据记录数量
total_count_array=(${total_count_query_result//|/})
total_count=${total_count_array[3]}

# 执行全表空值数量记录数量SQL查询语句
echo "----------检验全表空值记录数----------"
total_null_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select count(*) from $database_name.$table_name $null_where_sql_str"`
# 取出全量空值数据记录数量
null_count_array=(${total_null_query_result//|/})
null_count=${null_count_array[3]}

# 执行全表重复值数量记录数量SQL查询语句
echo "----------检验全表重复值记录数----------"
table_duplicate_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select sum(tmp.duplicate_count) as duplicate_sum from(select count(*) as duplicate_count from $database_name.$table_name group by $null_column having count(*)>1) as tmp"`
# 取出当日全量数据记录数量
duplicate_count_array=(${table_duplicate_query_result//|/})
duplicate_count=${duplicate_count_array[3]}

# 将所有数据校验结果插入到表中
echo "----------开始插入数据----------"
beeline -u jdbc:hive2://node01:10000 -n hdfs -e "insert into datacheck.table_total_quantity values($do_date,'$database_name','$table_name','increment_table',$null_count,$duplicate_count,$total_count)"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
#!/bin/bash
# 校验数据的库名
database_name=$1
# 校验数据的表名
table_name=$2
# 需要校验空值的列名，以 ‘,’分割
null_column=$3
# 日期
do_date=$4


# 初始化SQL查询语句
null_where_sql_str=''
# 将空值校验字段切分为数组
array=(${null_column//,/ })
# 遍历数组，拼接空值查询条件
for(( i=0;i< ${#array[@]};i++)) do
    if [ $i -eq 0 ];then
        null_where_sql_str="where ${array[i]} is null"
    else
        null_where_sql_str="$null_where_sql_str or ${array[i]} is null"
    fi
done;

# 执行当日全表数量记录数量SQL查询语句
echo "----------检验当日全量记录数----------"
total_count_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select count(*) from $database_name.$table_name"`
# 取出当日全量数据记录数量
total_count_array=(${total_count_query_result//|/})
total_count=${total_count_array[3]}

# 执行全表空值数量记录数量SQL查询语句
echo "----------检验全表空值记录数----------"
total_null_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select count(*) from $database_name.$table_name $null_where_sql_str"`
# 取出全量空值数据记录数量
null_count_array=(${total_null_query_result//|/})
null_count=${null_count_array[3]}

# 执行全表重复值数量记录数量SQL查询语句
echo "----------检验全表重复值记录数----------"
table_duplicate_query_result=`beeline -u jdbc:hive2://node01:10000 -n hdfs -e "select sum(tmp.duplicate_count) as duplicate_sum from(select count(*) as duplicate_count from $database_name.$table_name group by $null_column having count(*)>1) as tmp"`
# 取出当日全量数据记录数量
duplicate_count_array=(${table_duplicate_query_result//|/})
duplicate_count=${duplicate_count_array[3]}

# 将所有数据校验结果插入到表中
echo "----------开始插入数据----------"
beeline -u jdbc:hive2://node01:10000 -n hdfs -e "insert into datacheck.table_total_quantity values($do_date,'$database_name','$table_name','increment_table',$null_count,$duplicate_count,$total_count)"

四个参数：

第一个参数是数据库名
第二个参数是表名
第三个参数是字段名，字段名以逗号分隔
第四个参数是日期

ODS层数据质量校验

首先把ODS层中的表分为需要增量检查的表与需要全量检查的表，然后调用相应的脚本。

增量检查

订单详情表(ods_order_detail)
用户表(ods_user_info)
支付流水表(ods_payment_info)
。。。

全量检查

订单表(ods_order_info)
SKU商品表(ods_sku_info)
商品一级分类表(ods_base_category1)
商品二级分类表(ods_base_category2)
商品三级分类表(ods_base_category3)
。。。

这样的话，我们可以写一个shell脚本，将质量检测脚本调用每个表的语句都写进去。

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#!/bin/bash
# 增量检测

# 分区
data_date = $1


/usr/hdk/table_increment.sh mall $data_date ods_order_detail id,order_id,user_id,sku_id,sku_name,order_price,sku_num,create_time

/usr/hdk/table_increment.sh mall $data_date ods_user_info id,name,birthday,gender,email,user_level,create_time,operate_time

/usr/hdk/table_increment.sh mall $data_date ods_payment_info id,out_trado_no,order_id,user_id,alipay_trado_no,total-amount，subject，payment_type,payment_time

# 全量检测
/usr/hdk/table_total_quantity.sh mall ods_order_info  ..,..

...

然后就不用一个一个敲了，直接跑脚本就行。

DWD层数据质量校验

复制代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
create table dwd_table_data_check(
    data_date string comment '数据时间分区dt',
    source_database_name string comment '数据源库库名',
    source_table_name string comment '数据源表表名',
    source_column string comment '数据源表字段名',
    target_database_name string comment '数据目标库库名',
    target_table_name string comment '数据目标表表名',
    target_column string comment '数据目标表字段名',
    consistent_data_count bigint comment '全表数据一致记录数',
    source_table_count bigint comment '数据源表全表记录数', 
    target_table_count bigint comment '数据目标表全表记录数', 
    target_duplicate_count bigint  comment '数据目标表重复值记录数'
);

数据校验脚本：

复制代码

#!/bin/bash

# 数据源库名
source_database_name = $1

# 数据目标库名
target_database_name = $2

# 增量数据所在的日期分区
do_data=$3

# 校验数据源表的表名
source_table_name=$4

# 校验数据源表的字段（与目标表顺序一致才能对比两个字段）
source_column=$5

# 校验数据目标表的表名
target_table_name=$6

# 校验数据目标表的字段（与源表顺序一致才能对比两个字段）
target_column=$7

# 初始化SQL查询语句
join_on_sql_str=''

# 将校验数据源表的字段切成列名数组
source_column_array=(${source_column//,/})

# 将校验数据目标表的字段切成列名数组
target_column_array=(${target_column//,/})

# 遍历数组，拼接表关联关系，输入字段全部关联
for(( i=0;i< ${#source_column_array[@]};i++)) do
    if [$i -eq 0];then
        join_on_sql_str = "on $source_table_name.${source_column_array[i]}=$target_table_name.${target_column_array[i]}"
    else
        join_on_sql_str = "$join_on_sql_str and $source_table_name.${source_column_array[i]}=$target_table_name.${target_column_array[i]}"
    fi
done;

echo "----------ods-dwd 一致性检查----------"
# 执行数据源表和目标表的关联查询SQL语句，查询数据一致的条数
consistent_data_query_result=`hive -e "select count(*) from $source_database_name .$source_table_name join $target_database_name .$target_table_name $join_on_sql_str"`

# 取出全表查询数据一致的条数
consistent_data_cpunt=${consistent_data_query_result:3}

echo "----------ods层记录条数----------"
# 执行查询数据源表的记录条数
source_data_query_result=`hive -e "select count(*) from $source_database_name.$source_table_name"`

# 取出全表数据源表的记录条数
source_table_count=${source_data_query_result:3}

echo "----------dwd层记录条数----------"
# 执行查询数据目标表的记录条数
target_data_query_result=`hive -e "select count(*) from $target_database_name.$target_table_name"`

# 取出全表数据目标表的记录条数
target_table_count=${target_data_query_result:3}

# 执行全表重复值的记录数量SQL查询语句
table_duplicate_query_result=`hive -e "select sum(tmp.duplicate_count) as duplicate_sun from (select count(*) as duplicate_count from $target_database_name.$target_table_name group by $target_column having count(*)>1) as tmp"`

# 取出全表重复值数据记录数量
duplicate_count = ${table_duplicate_query_result:3}

# 将所有的数据检验结果插入表中
hive -e "insert into datacheck.dwd_table_data_check values('$do_date','$source_database_name ','$source_table_name','$source_column','$target_database_name','$target_table_name','$target_column','$source_table_count','$target_table_count','$duplicate_count')"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
#!/bin/bash

# 数据源库名
source_database_name = $1

# 数据目标库名
target_database_name = $2

# 增量数据所在的日期分区
do_data=$3

# 校验数据源表的表名
source_table_name=$4

# 校验数据源表的字段（与目标表顺序一致才能对比两个字段）
source_column=$5

# 校验数据目标表的表名
target_table_name=$6

# 校验数据目标表的字段（与源表顺序一致才能对比两个字段）
target_column=$7

# 初始化SQL查询语句
join_on_sql_str=''

# 将校验数据源表的字段切成列名数组
source_column_array=(${source_column//,/})

# 将校验数据目标表的字段切成列名数组
target_column_array=(${target_column//,/})

# 遍历数组，拼接表关联关系，输入字段全部关联
for(( i=0;i< ${#source_column_array[@]};i++)) do
    if [$i -eq 0];then
        join_on_sql_str = "on $source_table_name.${source_column_array[i]}=$target_table_name.${target_column_array[i]}"
    else
        join_on_sql_str = "$join_on_sql_str and $source_table_name.${source_column_array[i]}=$target_table_name.${target_column_array[i]}"
    fi
done;

echo "----------ods-dwd 一致性检查----------"
# 执行数据源表和目标表的关联查询SQL语句，查询数据一致的条数
consistent_data_query_result=`hive -e "select count(*) from $source_database_name .$source_table_name join $target_database_name .$target_table_name $join_on_sql_str"`

# 取出全表查询数据一致的条数
consistent_data_cpunt=${consistent_data_query_result:3}

echo "----------ods层记录条数----------"
# 执行查询数据源表的记录条数
source_data_query_result=`hive -e "select count(*) from $source_database_name.$source_table_name"`

# 取出全表数据源表的记录条数
source_table_count=${source_data_query_result:3}

echo "----------dwd层记录条数----------"
# 执行查询数据目标表的记录条数
target_data_query_result=`hive -e "select count(*) from $target_database_name.$target_table_name"`

# 取出全表数据目标表的记录条数
target_table_count=${target_data_query_result:3}

# 执行全表重复值的记录数量SQL查询语句
table_duplicate_query_result=`hive -e "select sum(tmp.duplicate_count) as duplicate_sun from (select count(*) as duplicate_count from $target_database_name.$target_table_name group by $target_column having count(*)>1) as tmp"`

# 取出全表重复值数据记录数量
duplicate_count = ${table_duplicate_query_result:3}

# 将所有的数据检验结果插入表中
hive -e "insert into datacheck.dwd_table_data_check values('$do_date','$source_database_name ','$source_table_name','$source_column','$target_database_name','$target_table_name','$target_column','$source_table_count','$target_table_count','$duplicate_count')"

7个参数：

第一个参数是数据源库名
第二个参数是数据目标库名
第三个参数是时间分区参数
第四个参数是需要进行数据校验的源表表名
第五个参数是需要进行数据校验的源表的字段，用’,'分隔
第六个参数是需要进行数据校验的目标表表名
第七个参数是需要进行数据校验的目标表的字段，用’,'分隔

最后

以上就是沉静蚂蚁最近收集整理的关于数据质量概述及通用脚本增量数据质量检验通用脚本（指定分区数据质量检验）全量数据质量检验通用脚本ODS层数据质量校验DWD层数据质量校验的全部内容，更多相关数据质量概述及通用脚本增量数据质量检验通用脚本（指定分区数据质量检验）全量数据质量检验通用脚本ODS层数据质量校验DWD层数据质量校验内容请搜索靠谱客的其他文章。

本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。

本文分类：# ---数据资产管理
浏览次数：74 次浏览
发布日期：2023-10-04 00:16:20
本文链接：https://www.kaopuke.com/article/k-p-k_14_uzogf2_14__7__10_w.html

数据质量概述及通用脚本增量数据质量检验通用脚本（指定分区数据质量检验）全量数据质量检验通用脚本ODS层数据质量校验DWD层数据质量校验

概述

数据质量标准分类

增量数据质量检验通用脚本（指定分区数据质量检验）

表格式：

脚本：

全量数据质量检验通用脚本

表格式：

脚本：

ODS层数据质量校验

增量检查

全量检查

DWD层数据质量校验

最后

评论列表共有 0 条评论

发表评论取消回复

数据质量概述及通用脚本增量数据质量检验通用脚本（指定分区数据质量检验）全量数据质量检验通用脚本ODS层数据质量校验DWD层数据质量校验

概述

数据质量标准分类

增量数据质量检验通用脚本（指定分区数据质量检验）

表格式：

脚本：

全量数据质量检验通用脚本

表格式：

脚本：

ODS层数据质量校验

增量检查

全量检查

DWD层数据质量校验

最后

相关文章

评论列表共有 0 条评论

发表评论 取消回复

微信扫一扫：分享

发表评论取消回复