hedfs和hive数据迁移后校验脚本

先谈论校验方法，本人腾讯云大数据工程师。
1、hdfs的校验

这个通常就是distcp校验，hdfs通过distcp迁移到另一个集群，怎么校验你的对不对。
有人会说，默认会有校验CRC校验。我们关闭了，为什么关闭？全量迁移，如果当前表再写数据，开自动校验就会失败。数据量大（PB级）迁移流程是先迁移全量，后面在定时补近来几天增量，再找个时间点，进行业务割接

那么怎么知道你迁移的hdfs是否有题目呢？
2个文件，一个是脚本，一个是必要校验的目录
data_checksum.py

1. # -*- coding: utf-8 -*-
2. # @Time    : 2025/1/16 22:52
3. # @Author  : fly-wlx
4. # @Email   : xxx@163.com
5. # @File    : data_compare.py
6. # @Software: PyCharm
8. import subprocess
11. #output_file = 'data_checksum_result.txt'
12. def load_file_paths_from_conf(conf_file):
13.     file_list = []
14.     with open(conf_file, 'r') as file:
15.         lines = file.readlines()
16.         for line in lines:
17.             path = line.strip()
18.             if path and not path.startswith('#'):  # 跳过空行和注释
19.                 full_path = f"{path}"
20.                 file_list.append(full_path)
21.     return file_list
23. #def write_sizes_to_file(filepath,source_namenode,source_checksum,target_namenode,target_checksum,status, output_file):
24. #    with open(output_file, 'w') as file:
25. #file.write(f"{source_namenode}/{filepath},{source_checksum},{target_namenode}/{filepath},{target_checksum},{status}\n")
27. def write_sizes_to_file(source_path, src_info, destination_path, target_info, status,output_file):
28.     with open(output_file, 'a') as file:
29.          file.write(f"{source_path},{src_info},{destination_path}, {target_info}, {status}\n")
30. def run_hadoop_command(command):
31.     """运行 Hadoop 命令并返回输出"""
32.     try:
33.         result = subprocess.check_output(command, shell=True, text=True)
34.         return result.strip()
35.     except subprocess.CalledProcessError as e:
36.         print(f"Command failed: {e}")
37.         return None
39. def get_hdfs_count(hdfs_filepath):
40.     """获取 HDFS 路径的文件和目录统计信息"""
41.     command = f"hadoop fs -count {hdfs_filepath}"
42.     output = run_hadoop_command(command)
43.     if output:
44.         parts = output.split()
45.         if len(parts) >= 3:
46.             dir_count, file_count, content_size = parts[-3:]
47.             return dir_count, file_count, content_size
48.     return None, None, None
50. def get_hdfs_size(hdfs_filepath):
51.     """获取 HDFS 路径的总文件大小"""
52.     command = f"hadoop fs -du -s {hdfs_filepath}"
53.     output = run_hadoop_command(command)
54.     if output:
55.         parts = output.split()
56.         if len(parts) >= 1:
57.             return parts[0]
58.     return None
60. def validate_hdfs_data(source_namenode, target_namenode,filepath):
61.     output_file = 'data_checksum_result.txt'
62.     source_path=f"{source_namenode}/{filepath}"
63.     destination_path = f"{target_namenode}/{filepath}"
64.     """校验 HDFS 源路径和目标路径的数据一致性"""
65.     print("Fetching source path statistics...")
66.     src_dir_count, src_file_count, src_content_size = get_hdfs_count(source_path)
67.     src_total_size = get_hdfs_size(source_path)
69.     print("Fetching destination path statistics...")
70.     dest_dir_count, dest_file_count, dest_content_size = get_hdfs_count(destination_path)
71.     dest_total_size = get_hdfs_size(destination_path)
72.     src_info={}
73.     src_info["src_dir_count"] = src_dir_count
74.     src_info["src_file_count"] = src_file_count
75.     #src_info["src_content_size"] = src_content_size
76.     src_info["src_total_size"] = src_total_size
77.     target_info = {}
78.     target_info["src_dir_count"] = dest_dir_count
79.     target_info["src_file_count"] = dest_file_count
80.     #target_info["src_content_size"] = dest_content_size
81.     target_info["src_total_size"] = dest_total_size
83.     print("\nValidation Results:")
84.     if (src_dir_count == dest_dir_count and
85.         src_file_count == dest_file_count and
86.        # src_content_size == dest_content_size and
87.         src_total_size == dest_total_size):
88.         print("✅ Source and destination paths are consistent!")
89.         write_sizes_to_file(source_path, src_info, destination_path,target_info, 0,
90.                             output_file)
91.     else:
92.         print("❌ Source and destination paths are inconsistent!")
93.         write_sizes_to_file(source_path, src_info, destination_path, target_info, 1,
94.                             output_file)
95.         #print(f"Source: DIR_COUNT={src_dir_count}, FILE_COUNT={src_file_count}, CONTENT_SIZE={src_content_size}, TOTAL_SIZE={src_total_size}")
96.         #print(f"Destination: DIR_COUNT={dest_dir_count}, FILE_COUNT={dest_file_count}, CONTENT_SIZE={dest_content_size}, TOTAL_SIZE={dest_total_size}")
98. # 设置源路径和目标路径
99. #source_path = "hdfs://namenode1:8020/"
100. #destination_path = "hdfs://namenode2:8020/path/to/destination"
101. # 定义源和目标集群的 namenode 地址
102. source_namenode = "hdfs://10.xx.xx.6:8020"
103. target_namenode= "hdfs://10.xx.xx.106:4007"
105. def main():
106.     # 配置文件路径和输出文件路径
107.     conf_file = 'distcp_paths.conf'
108.     # 定义源和目标集群的 namenode 地址
110.     # 设置源路径和目标路径
111.     #source_namenode = "hdfs://source-namenode:8020"
112.     #target_namenode = "hdfs://target-namenode:8020"
114.     # 文件列表
115.     file_paths = load_file_paths_from_conf(conf_file)
117.     # 对每个目录进行校验
118.     for filepath in file_paths:
119.         validate_hdfs_data(source_namenode, target_namenode, filepath)
122.    
125. if __name__ == "__main__":
126.     main()
128. # 执行校验
129. #validate_hdfs_data(source_path, destination_path)

复制代码

distcp_paths.conf

1. /apps/hive/warehouse/xx.db/dws_ixx_features
2. /apps/hive/warehouse/xx.db/dwd_xx_df

复制代码

用法

直接python3 data_checksum.py（必要改为自己的）
他会实时打印对比结果，并且将结果生成到一个文件中（data_checksum_result.txt）

2、hive文件内容比对

终极客户要的是任务的数据对得上，而不是管你迁移怎么样，所以验证任务的方式：双方同时跑同多个Hive任务流的任务，检察表数据内容是否同等。（因为跑出来的hdfs的文件大小由于mapreduce原因，肯定是不同等的，校验实际数据同等就行了）
方法是先对比表字段，然后对比count数，然后将每行拼起来对比md5
涉及3个文件，单检测脚本，批量入口脚本，必要批量检测的表文件

check_script.sh

1. #!/bin/bash
2. #owner:clark.shi
3. #date:2025/1/22
4. #背景:用于hive从源端任务和目标端任务，两边跑完结果表的内容校验（因为mapreduce和小文件不同，所以要用数据内容校验）
5. #     --用trino(presto）会更好,因为可以跨集群使用，目前客户因为资源情况没装，此为使用hive引擎，将数据放到本地进行比对
8. #输入：源端表，目标表，分区名，分区值
9. #$0是脚本本身，最低从1开始
12. #限制脚本运行内存大小,30gb
13. #ulimit -v 30485760
15. #---注意，要保证，2个表的字段顺序是一样的（md5是根据顺序拼接的）
16. echo "================"
17. echo "注意"
18. echo "要保证，2个表的字段顺序是一样的（md5是根据顺序拼接的)"
19. echo "要保证，这2个表是存在的"
20. echo "要保证，双端是可以互相访问"
21. echo "要保证，2个hive集群的MD5算法相同"
22. echo "禁止表，一个分区数据量超过本地磁盘，此脚本会写入本地磁盘（双端数据），对比后删除"
23. echo "注意，如果分区字段是数字不用加引号，如果是字符串需要加引号,搜partition_value,这里分区是int如20250122是没有引号"
24. echo "================"
26. a_table=$1
27. b_table=$2
28. partition_column=$3
29. partition_value=$4
32. if [ $# -ne 4 ]; then
33.     echo "错误：必须输入 4 个参数,源端表，目标表，分区名，分区值"
34.     exit 1
35. fi
37. #------------函数
39. check_value() {
40.     # 第一个参数是布尔值，第二个参数是要 echo 的内容
41.     local value=$1
42.     local message=$2
43.    
44.     # 检查第一个参数的值
45.     if [ "$value" == "false" ]; then
46.         echo "校验失败：$message" >> rs.txt
47.         exit
48.     fi
49. }
53. #-----------函数结束
56. echo "需要对比表的数据内容是$a_table和$b_table--，需要对比分区$partition_column是$partition_value--"
58. sleep 2
59. echo "===============开始校验============="
60. #todo改成自己的，kerbers互信认证(也可以用ldap)
61. `kinit -kt /root/s_xx_tbds.keytab s_xx_tbds@TBDS-V12X10CS`
64. #校验字段类型
65. echo "1.开始校验字段类型"
67.        
68. #todo这里要改成自己的
69.   beeline -u "jdbc:hive2://10.xx.xx.4:10001/XXdatabase;principal=hive/tbds-10-xx-xx-4.hadooppdt.xxjin.srv@TBDS-V12X10CS;transportMode=http;httpPath=cliservice" -e "DESCRIBE $b_table" > 1_a_column.txt
70.   beeline -u "jdbc:hive2://10.xx.xx.104:7001/XXdatabase;principal=hadoop/10.xx.xx.104@TBDS-09T7KXLE" -e "DESCRIBE $a_table" > 1_b_column.txt
71.   if diff 1_a_column.txt 1_b_column.txt > /dev/null; then
72.     echo "表结构一致"
73.   else
74.     echo "表结构不一致"
75.     check_value false "$a_table和$b_table字段类型不一致"
76.   fi
79. echo "------------1.表字段，校验完毕,通过-------------"
82. #校验count数
83. echo "2.开始count校验"
84.   beeline -u "jdbc:hive2://10.xx.xx.4:10001/XXdatabase;principal=hive/tbds-10-xx-xx-4.hadooppdt.xxjin.srv@TBDS-V12X10CS;transportMode=http;httpPath=cliservice" -e "select count(*) from $b_table where $partition_column=$partition_value" > 2_a_count.txt
85.     beeline -u "jdbc:hive2://10.xx.xx.104:7001/XXdatabase;principal=hadoop/10.xx.xx.104@TBDS-09T7KXLE" -e "select count(*) from $a_table where $partition_column=$partition_value" > 2_b_count.txt
86.   if diff 2_a_count.txt 2_b_count.txt > /dev/null; then
87.     echo "数据行一致"
88.   else
89.     echo "数据行不一致"
90.     check_value false "$a_table和$b_table的数据行不一致"
91.   fi
93. echo "------------2.数据行，校验完毕,通过-------------"
95. #拼接每一行的值，作为唯一值，创建2个临时表
96. echo "3.生成每条数据唯一标识"
97.   #1.获取表列名
98.   #使用awk，去除第一行字段名，，删除#字号以及他后面的内容（一般是分区的描述），根据分隔符|取第一列数据，去掉空的行
99.   beeline -u "jdbc:hive2://10.xx.xx.104:7001/XXdatabase;principal=hadoop/10.xx.xx.104@TBDS-09T7KXLE" --outputformat=dsv -e "DESCRIBE $a_table" |awk 'NR > 1' |awk '!/^#/ {print} /^#/ {exit}'|awk 'BEGIN {FS="|"} {print $1}'|awk 'NF > 0' > 3_table_field_name.txt
100.   #2.拼接表列名，生成md5的表 （第一步已经检测过双方的表结构了，这里用同一个拼接字段即可）
102.   # 使用 while 循环逐行读取文件内容
103.   name_fields=""
104.   while IFS= read -r line; do
105.     if [ -z "$name_fields" ]; then
106.       name_fields="$line"
107.     else
108.       name_fields="$name_fields,$line"
109.     fi
110.   done < "3_table_field_name.txt"
111.   echo "$name_fields"
112.   #将每行数据进行拼接，并且生成含一个字段的md5表
113.   md5_sql="SELECT distinct(MD5(CONCAT($name_fields))) AS md5_value "
114.   a_md5_sql="$md5_sql from (select * from dim_user_profile_df where $partition_column=$partition_value  limit 100)a;"
115.   b_md5_sql="$md5_sql from $a_table where $partition_column=$partition_value;"
116.   echo "a表的sql是:$a_md5_sql"
117.   echo "b表的sql是:$b_md5_sql"
119.   #源端是生产环境，这里做了特殊处理，源端就取100条(没使用order by rand()，客户主要是检测函数,order by 会占用他们集群资源)
120.   beeline -u "jdbc:hive2://10.xx.xx.4:10001/XXdatabase;principal=hive/tbds-10-xx-xx-4.hadooppdt.xxjin.srv@TBDS-V12X10CS;transportMode=http;httpPath=cliservice" --outputformat=dsv -e "$a_md5_sql" > 4_a_md5_data.txt
121.   beeline -u "jdbc:hive2://10.xx.xx.104:7001/XXdatabase;principal=hadoop/10.xx.xx.104@TBDS-09T7KXLE" --outputformat=dsv -e "$b_md5_sql" > 4_b_md5_data.txt
123.   #3.（由于不是同集群，需要下载到本地，再进行导入--如果耗费资源时长太长，再导入到hive，否则直接shell脚本搞定）
124.   # 设置large_file和small_file的路径
125.   large_file="4_b_md5_data.txt"
126.   small_file="4_a_md5_data.txt"
127.   # 遍历small_file中的每一行
128.   while IFS= read -r line; do
129.       # 检查line是否存在于large_file中
130.       if grep -qxF "$line" "$large_file"; then
131.           # 如果line存在于large_file中，输出1
132.           #echo "1"
133.           a=1
134.       else
135.           # 如果line不存在于large_file中，输出2
136.           echo "2"
137.           check_value false "$a_table和$b_table抽样存在数据内容不一致"
138.       fi
139.   done < "$small_file"
141.   echo echo "------------3.数据内容，校验完毕,通过-------------"
142. #抽样核对md5（取数据时已抽样，否则数据太大容易跑挂生产环境）

复制代码

input_file.txt必要校验的表文件

源端表名，目标端表名，分区字段（写1级分区就可以），分区值
   ods_xxnfo_di ods_xxnfo_dii dt 20250106
  ods_asxx_log_di ods_asxx_log_dii dt 20250106
  ods_xxog_di ods_xxog_di dt 20250106
  dwd_xxx dwd_xxx dt 20250106
  run.sh

1. #!/bin/bash
3. # 设置文件路径
4. input_file="input_file.txt"
6. # 遍历文件中的每一行
7. while IFS= read -r line; do
8.     # 调用另一个脚本并传递当前行的参数
9.     echo $line
10.     ./check_script.sh $line
11.     # 在每次执行完后间隔一小段时间，避免系统过载（可选）
12.     sleep 1
13. done < "$input_file"

复制代码

使用方法

sh run.sh（必要把check_scripe和run里的内容改成自己的哈）
他会把不通过的，生成一个rs.txt

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