首页 > 数据库 >Python-批量统计MySQL中表的数据量

Python-批量统计MySQL中表的数据量

时间:2024-08-23 10:04:47浏览次数:12  
标签:count str Python 中表 print 数据量 date table data

背景

在数据中台中,有时为了核对数据,需要每天批量统计MySQL数据库中表的数据量,但是DMS中没有周期调度功能。

MySQL创建表

--统计的表清单
CREATE TABLE `dws_table_list` (
  `table_name` varchar(255) DEFAULT NULL,
  `flag` varchar(255) DEFAULT NULL
);

--每天的数据量
CREATE TABLE `dws_table_count` (
  `table_name` varchar(255)  DEFAULT NULL,
  `table_count` bigint DEFAULT NULL,
  `ds` varchar(255) DEFAULT NULL
); 

dws_table_list表中插入数据

说明:test_table1、test_table2、test_table3这些表,需要具有相同的日期字段(比如:ds)。

Python代码

import pymysql
import sys
from datetime import datetime, timedelta

if __name__ == '__main__':

    if len(sys.argv) == 2:
        input_str = sys.argv[1]
    else:
        print("参数个数不对!")
        exit()

    # MySQL中表不存在的列表
    table_not_exists_list = []
    # 没有业务数据日期的列表
    dataDate_no_data_list = []

    # 业务数据日期
    data_date = datetime.strptime(input_str, "%Y%m%d")
    data_date_str = data_date.strftime("%Y%m%d")
    print(f"业务数据日期: {data_date_str}")

    # 日期减去特定的时间间隔
    data_date_1days_ago_str = (data_date - timedelta(days=1)).strftime('%Y%m%d')
    data_date_2days_ago_str = (data_date - timedelta(days=2)).strftime('%Y%m%d')
    data_date_3days_ago_str = (data_date - timedelta(days=3)).strftime('%Y%m%d')
    data_date_4days_ago_str = (data_date - timedelta(days=4)).strftime('%Y%m%d')
    data_date_5days_ago_str = (data_date - timedelta(days=5)).strftime('%Y%m%d')
    data_date_6days_ago_str = (data_date - timedelta(days=6)).strftime('%Y%m%d')
    data_date_7days_ago_str = (data_date - timedelta(days=7)).strftime('%Y%m%d')

    # 连接MySQL数据库
    config = {
        'user': 'root',
        'password': 'Root@1234',
        'host': 'localhost',
        'database': 'test'
    }
    conn = pymysql.connect(**config)

    # 创建一个游标对象
    cursor = conn.cursor()

    # 执行一个查询
    query = "SELECT  table_name  from  dws_table_list ORDER BY table_name ;"
    cursor.execute(query)

    # 打印表清单中的所有表的数量
    print("dws_table_list表中的表个数: ",cursor.rowcount)
    print("**"*100)

    # 可取出指针结果集中的所有行,返回的结果集一个元组。
    result = cursor.fetchall()

    # 删除业务日期当天的数据,防止重跑时数据重复
    cursor.execute("DELETE FROM dws_table_count WHERE ds = %s ;", data_date_str)

    # 遍历指针结果集
    for rows in result:
        table_name = rows[0]
        query_count = f"SELECT  '{table_name}' table_name,COUNT(0),ds table_count FROM {table_name} WHERE ds = '{data_date_str}' GROUP BY ds ;"
        print(query_count)
        try:
            cursor.execute(query_count)
            result2 = cursor.fetchall()

            if cursor.rowcount > 0:
                # 插入数据
                insert_sql = "INSERT INTO dws_table_count VALUES (%s,%s,%s)"
                for rows2 in result2:
                    cursor.execute(insert_sql, rows2)
            else:
                print("\033[93m" + f"Table {table_name} does not have data for {data_date_str} " + "\033[0m")
                dataDate_no_data_list.append(table_name)

        except pymysql.MySQLError as e:
            if "exist" in str(e):
                print("\033[31m" + "MySQL error: " + str(e) + "\033[0m")
                table_not_exists_list.append(table_name)
            else:
                print("\033[31m" + "MySQL error: " + str(e) + "\033[0m")
                exit()

    # 删除7天之前的数据
    cursor.execute("DELETE FROM dws_table_count WHERE ds = %s ;",data_date_7days_ago_str)

    # 根据表中的数据展示需要的结果
    final_query = (f"""SELECT  
                        table_name,
                        '{data_date_str}'  ds,
                        SUM(IF(ds = '{data_date_str}',table_count,0))  data_{data_date_str},
                        SUM(IF(ds = '{data_date_1days_ago_str}',table_count,0))  data_{data_date_1days_ago_str},
                        SUM(IF(ds = '{data_date_2days_ago_str}',table_count,0))  data_{data_date_2days_ago_str},
                        SUM(IF(ds = '{data_date_3days_ago_str}',table_count,0))  data_{data_date_3days_ago_str},
                        SUM(IF(ds = '{data_date_4days_ago_str}',table_count,0))  data_{data_date_4days_ago_str},
                        SUM(IF(ds = '{data_date_5days_ago_str}',table_count,0))  data_{data_date_5days_ago_str},
                        SUM(IF(ds = '{data_date_6days_ago_str}',table_count,0))  data_{data_date_6days_ago_str}
                        FROM  dws_table_count 
                        WHERE ds  BETWEEN  '{data_date_6days_ago_str}'  AND  '{data_date_str}'
                        GROUP BY table_name ;
                    """)

    cursor.execute(final_query)
    # 可取出指针结果集中的所有行,返回的结果集一个元组。
    result3 = cursor.fetchall()

    print("**"*100)
    print("\033[92m" + "MySQL表数据量统计:" + "\033[0m")
    print("--"*60)
    # 获取列名
    columns = [column[0] for column in cursor.description]
    # 打印列名
    print('xh',end='|')
    for col in columns:
        print(col,end='|')
    print()

    # 打印结果
    count = 1
    for rows3 in result3:
        print(count,end='|')
        for j in range(len(rows3)):
            print(rows3[j],end='|')
        count = count + 1
        print()
    print("**" * 100)

    print("\033[92m" + "MySQL中不存在的表:" + "\033[0m")
    print("--"*60)
    for index,table in enumerate(table_not_exists_list,start=1):
        print(index,table,sep='|')
    print("**" * 100)
    print(f"\033[92m" + f"业务日期{data_date_str}没有数据的表: " + "\033[0m")
    print("--"*60)
    for index,table in enumerate(dataDate_no_data_list,start=1):
        print(index,table,sep='|')

    # 关闭游标和连接
    cursor.close()
    conn.commit()
    conn.close()

执行结果

 

标签:count,str,Python,中表,print,数据量,date,table,data
From: https://www.cnblogs.com/yeyuzhuanjia/p/18375375

相关文章

  • python socket编辑示例
    服务端代码:fromsocketimportsocket,AF_INET,SOCK_STREAM#1.创建socket对象AF_INET:用于internet之间的进程通信,SOCK_STREAM:表示TCP协议server_socket=socket(AF_INET,SOCK_STREAM)#2.绑定ip和端口号ip='127.0.0.1'port=8888server_socket.bind((ip,p......
  • 基于Python flask的图书借阅管理系统的设计与实现
    基于PythonFlask的图书借阅管理系统旨在为图书馆或类似机构提供一个高效、便捷的管理平台,覆盖图书借阅的各个环节,帮助管理员和读者更好地管理和使用图书资源。该系统采用Python编程语言和Flask框架进行开发,结合了数据库管理、用户认证、数据可视化等技术,确保系统的功能完备和......
  • python-jose 实现fastapi登录验证
    JWT和Session的区别:JWT:JWT是一种无状态的认证机制。由于JWT令牌包含了用户的身份信息以及相关的元数据,服务端不需要存储任何用户状态信息,只需要验证JWT令牌的真实性和有效性即可。这使得JWT非常适合于构建无状态的分布式系统,因为JWT令牌可以在不同的服务之间轻松共享。Sessio......
  • Python正则表达式提取车牌号
    在Python中使用正则表达式(RegularExpressions)来提取车牌号是一个常见的任务,尤其是在处理车辆信息或进行图像识别后的文本处理时。中国的车牌号格式多种多样,但通常包含省份简称、英文字母和数字。以下是一个使用Python正则表达式来提取常见中国车牌号(包括新能源车牌)的示例代码。1......
  • 基于 Python 3.12.4 和 PySide 6 的 假用户操作系统【0-0】虚拟终端框架
    基本目标实现工作目录的切换;实现非退出命令下的无限循环;到目前为止还不能实现的功能动态导入py文件中的模块缺少组件定义虚拟进程类command.pyfromctypesimportwindllfromPySide6.QtWidgetsimportQWidgetfromsysimportexitassys_exitfromosimportpath......
  • python怎么用正则表达式筛选网页内容
    一、正则表达式简述:什么是正则表达式?正则表达式就是可以匹配文本片段的模式,最简单的正则表达式就是一个字符串,用于在文本中匹配到此字符串自身。二、常用正则表达式:设计正则表达式的时候有几个注意点如下:a.特殊符号需要加转移符:如要匹配'china.com',则正则表达式格式应为'......
  • python中%s是什么
    python的%s是用作print的格式化输出:打印字符串。Python中内置的%操作符可用于格式化字符串操作,控制字符串的呈现格式,s代表字符串。格式符为真实值预留位置,并控制显示的格式。格式符可以包含有一个类型码,用以控制显示的类型。python中不同类型的字符拥有不同的格式符,分别为......
  • python 05-标准库:csv、json、sqlite3、datetime模块
    csv模块importcsv#data.csv不存在时,会现在当前目录下创建一个data.csv文件withopen("data.csv","w",encoding='utf-8')asfile:writer=csv.writer(file)writer.writerow(["trasanction_id","product_id","pric......
  • Selenium + Python 自动化测试22(PO+数据驱动)
            我们的目标是:按照这一套资料学习下来,大家可以独立完成自动化测试的任务。上一篇我们讨论了PO模式和unittest框架结合起来使用。        本篇文章我们综合一下之前学习的内容,如先将PO模式、数据驱动思想和我们生成HTML报告融合起来,综合的灵活的使用......
  • 第四章 Python操作redis(操作案例)
    一、python对redis基本操作(1)连接redis#方式1importredisr=redis.Redis(host='127.0.0.1',port=6379)r.set('foo','Bar')print(r.get('foo'))#方式2importredispool=redis.ConnectionPool(host='127.0.0.1',po......