Python脚本之根据excel统计表中字段值的缺失率实用案例

缺失率 = （该字段NULL值+NA值+空字符串 的记录数）/该表总记录数

1. 将需要统计的表名和字段以及类型放在excel里边；
2. 使用 pandas 读取excel的数据；
3. 连接数据库；
4. 将读取到excel里边的数据拼接如sql里边统计；
5. 将计算结果写回到 excel 中。

import pymssql
import pandas as pd
import csv


def get_pandas_data():
    df = pd.read_excel(r'C:\Users\lucha\Desktop\luchangyin.xlsx',header=None)
    data_list = [(df.iloc[i,0],df.iloc[i,1],df.iloc[i,2]) for i in df.index.values]
    return data_list

def get_sqlserver_data():
    # 定义要写入的目标csv文件
    f = open(r'C:\Users\lucha\Desktop\wuxuan.csv', "w", newline='')
    writer = csv.writer(f)
    writer.writerow(['errorCounts', 'total', 'TableName', 'TableColumn', '错误率'])
    # 打开数据库连接 这里的host='.'也可用本机ip或ip+端口号(sqlserver默认端口号：1433)
    conn = pymssql.connect(host="ip", user="username", password="password",
                           database="dbname", port="1433", charset='utf8')
    # 使用cursor()方法获取操作游标
    cursor = conn.cursor()
    data_list = get_pandas_data()
    for d_tuple in data_list:
        # SQL 查询语句
        tableName = d_tuple[0]
        tbColumn = d_tuple[1]
        typeColum = d_tuple[2] # 获取类型

        if typeColum == 'varchar' or typeColum == 'nvarchar' or typeColum == 'char' :
            sql = "select errorCounts,total from (select count(*) total,sum(case when " + tbColumn + " IS NULL or "+ tbColumn +"='' or "+ tbColumn +"='NULL' then 1 else 0 end) errorCounts from dbo." + tableName + ") a;"
        else:
            sql = "select errorCounts,total from (select count(*) total,sum(case when " + tbColumn + " IS NULL then 1 else 0 end) errorCounts from dbo." + tableName + ") a;"
        try:
            # 执行SQL语句
            cursor.execute(sql)
            # 获取所有记录列表
            results = cursor.fetchall()
            errorCounts = results[0][0]
            total = results[0][1]
            # print("errorCounts: "+ str(errorCounts), "\ttotal: "+ str(total), "\t\t"+ tableName+"."+ tbColumn +": "+ str(round((errorCounts/total) * 100,2)) +"%")
            persent = str(round((errorCounts/total) * 100,2)) +"%"
            # 保存为csv文件
            dbnote_list = [errorCounts, total, tableName, tbColumn, persent]
            print("管道文件测试-> %s" % dbnote_list)  # 输出测试
            writer.writerow(dbnote_list)
        except:
            print("报错咯。。。",results)
    # 关闭数据库连接
    conn.close()
    f.close()

# 程序入口
if __name__ == '__main__':
    get_sqlserver_data()