专栏首页AustinDatabases--如何用PYTHON 定时打印 MYSQL FREE 使用率,与自动创建测试数据库表

--如何用PYTHON 定时打印 MYSQL FREE 使用率,与自动创建测试数据库表

源数据库汇中,PYTHON 的使用不是一个可选项,主要在很多地方,监控,处理一些DEVOPS的事情,或者与业务有关的处理的工作都是需要PYTHON 来进行的。下面就用PTYHON 来完成一个很小的打印MYSQL 系统的内存占用率的小脚本来开始 PYTHON travel。(由于是初级水平有待提高,部分代码的有待进步)

在学习PYTHON 的过程中,(很菜)领会到PYTHON 本身的语法是一回事,你使用的各种包的熟悉又是另一回事。所以下面先得说说程序中使用的mysql 的 python connector.

PYTHON 连接到MYSQL 的包有很多 PYMYSQL , MYSQLAB, 这里没有使用而是使用了官方的 Connector/Python 的方式进行连接

下面相关的代码的初衷主要在分析一段时间INNODB BUFFER 的使用率,查看相关的变动情况,当然这样的监控也有各种图形化的监控软件,但灵活性不高

#!/usr/bin/env python3
# coding: utf-8
import mysql.connector
from mysql.connector import errorcode
import re
import time
import datetime
import sys


class DBFREEMEMORY:
    def __init__(self, user=None, passwd=None, host=None, db=None):
        self.user = user
        self.passwd = passwd
        self.host = host
        self.db = db


    def mysql_connect(self):
        remotedb = {
          'host': self.host ,
          'user': self.user,
          'passwd': self.passwd,
          'database': self.db,
          'charset': 'utf8',
          'connection_timeout': 30,
          'use_pure': True
        }

        try:
          connect = mysql.connector.connect(**remotedb)
          mycursor = connect.cursor(dictionary=True)
          sql = "select substring_index(substring_index(event_name,'/',2),'/',-1) as event_type, round(sum(current_number_of_bytes_used) / 1024/1024, 2) as MB_CURRENTLY_USED from performance_schema.memory_summary_global_by_event_name group by event_type having mb_currently_used >0"
          ) as event_type, round(sum(current_number_of_bytes_used) / 1024/1024, 2) as MB_CURRENTLY_USED from performance_schema.memory_summary_global_by_event_name group by event_type having mb_currently_used >0"
          mycursor.execute(sql)
memory = mycursor.fetchall()#收集当前使用的内存数
sql_1 = "show global variables  like 'innodb_buffer_pool_size';"
          mycursor.execute(sql_1)
          full_memory = mycursor.fetchall()
          #收集当前整体数据库占有的内存

          for t in full_memory:          if t['Value'] != None:
              fmem = float(t['Value']) / 1024 / 1024
            else:
              t['Value'] = 1
          for i in memory:
            if i['MB_CURRENTLY_USED'] != None:
               mem = i['MB_CURRENTLY_USED']
            else:
              i['MB_CURRENTLY_USED'] = 1
          result = format(float(mem) / float(fmem) * 100, '.2f')
          print(str(result) + '%' + '  ' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
          #将当前的内存使用数的百分比进行比较,并和当前时间一起打印

        except mysql.connector.Error as err:
          if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print("Something is wrong with your user name or password")
          elif err.errno == errorcode.ER_BAD_DB_ERROR:
            print("Database does not exist")
          else:
            print(err)
        finally:
          mycursor.close()
          connect.close()


if __name__ == '__main__':
   info = DBFREEMEMORY(user='admin', passwd='1234.Com', host='192.168.198.9', db='performance_schema')
   info.mysql_connect()
          'host': self.host ,
          'user': self.user,
          'passwd': self.passwd,
          'database': self.db,
          'charset': 'utf8',
          'connection_timeout': 30,
          'use_pure': True
        }

        try:
          connect = mysql.connector.connect(**remotedb)
          mycursor = connect.cursor(dictionary=True)

          sql = "select substring_index(substring_index(event_name,'/',2),'/',-1) as event_type, round(sum(current_number_of_bytes_used) / 1024/1024, 2) as MB_CURRENTLY_USED from performance_schema.memory_summary_global_by_event_name group by event_type having mb_currently_used >0"
          mycursor.execute(sql)
          memory = mycursor.fetchall()
          sql_1 = "show global variables  like 'innodb_buffer_pool_size';"
          mycursor.execute(sql_1)
          full_memory = mycursor.fetchall()

          for t in full_memory:
            if t['Value'] != None:
              fmem = float(t['Value']) / 1024 / 1024
            else:
              t['Value'] = 1
          for i in memory:
            if i['MB_CURRENTLY_USED'] != None:
               mem = i['MB_CURRENTLY_USED']
            else:
              i['MB_CURRENTLY_USED'] = 1
          result = format(float(mem) / float(fmem) * 100, '.2f')
          print(str(result) + '%' + '  ' + time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))


        except mysql.connector.Error as err:
          if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print("Something is wrong with your user name or password")
          elif err.errno == errorcode.ER_BAD_DB_ERROR:
            print("Database does not exist")
          else:
            print(err)


if __name__ == '__main__':
   info = DBFREEMEMORY(user='admin', passwd='1234.Com', host='192.168.198.9', db='performance_schema')
   info.mysql_connect()

下面一个程序是针对自动生成测试数据库表,下面会在数据库层面自动生成test 库 以及 test1表,并插入随机数 150万

#!/usr/bin/env python3
# coding: utf-8
from __future__ import print_function
import mysql.connector
from mysql.connector import errorcode
from datetime import date, datetime, timedelta
import re
import time
import datetime
import sys
import random


class DBFREEMEMORY:
    def __init__(self, user=None, passwd=None, host=None, db=None):
        self.user = user
        self.passwd = passwd
        self.host = host
        self.db = db

    def gen_random_string(self):  #产生随机内容的方法

        char_list = list('1234567890' + '0123456789')
        random.shuffle(char_list)
        return ''.join(char_list)

    def mysql_connect(self):
        remotedb = {
            'host': self.host,
            'user': self.user,
            'passwd': self.passwd,
            'database': self.db,
            'charset': 'utf8',
            'connection_timeout': 30,
            'use_pure': True
        }

        try:
            connect = mysql.connector.connect(**remotedb)
            mycursor = connect.cursor(dictionary=True)
            #判断当前的服务器是否已经存在test数据库
            mycursor.execute("show databases")
            database = [mycursor.fetchall()]
            # print (tables)
            database_list = re.findall('(\'.*?\')', str(database))
            database_list = [re.sub("'", '', each) for each in database_list]
            print(database_list)
            #如果存在test 数据库就直接退出
            if 'test' in database_list:
                print('The database of test has existed,it has deleted it,please run the job again')
               

            else:#创建相关                         mycursor.execute("create database test")          print('You have test database')

            DB_NAME = 'test'
            mycursor.execute("USE test".format(DB_NAME))
            #建表
            TABLES = {}
            TABLES['test'] = (
                "CREATE TABLE `test1` ("
                "  `id` int(11) NOT NULL AUTO_INCREMENT,"
                "  `content` varchar(200)  NULL,"
                "  `hash` varchar(200)  NULL,"
                "  `insert_date` date  NULL,"
                "  PRIMARY KEY (`id`)"
                ") ENGINE=InnoDB")

            table_name = TABLES['test']
            # mycursor.execute(table_name)
            mycursor.execute("show tables")
            table = [mycursor.fetchall()]
            # print (tables)
            table_list = re.findall('(\'.*?\')', str(table))
            table_list = [re.sub("'", '', each) for each in table_list]
            print(table_list)
            #判断如果没有
            if 'test1' in table_list:
                print('The table of test has existed,please delete it')

            else:
                try:  #执行并开始插入数据 10000条一提交
                    mycursor.execute(table_name)
                    #print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                    s_time = time.time()
                    i = 0
                    while i < 150000:
                        content = self.gen_random_string()
                        sql = "INSERT INTO test1 (content, hash,insert_date) VALUES ('%s', '%d',now())" \
                              % (content, hash(content))
                        mycursor.execute(sql)
                        i += 1
                        if i % 10000 == 0:
                           connect.commit()
                           print(i)
                    connect.close()
                    print('You have test table')
                    en_time = time.time()
                    print(en_time-s_time)
                    #print(time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
                except Exception as e:
                    print(e)





        except mysql.connector.Error as err:
            if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
                print("Something is wrong with your user name or password")
            elif err.errno == errorcode.ER_BAD_DB_ERROR:
                print("Database does not exist")
            else:
                print(err)
        finally:
            mycursor.close()
            connect.close()


if __name__ == '__main__':
    info = DBFREEMEMORY(user='admin', passwd='1234.Com', host='192.168.198.9', db='performance_schema')
info.mysql_connect()

本文分享自微信公众号 - AustinDatabases(AustinDatabases),作者:carol11

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2020-06-09

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • MYSQL 字符隐式转换,8 VS 57 ,新人哭旧人笑?

    周六特刊的那期SQL SERVER 你不仁,别怪他不义那期本来并未期望有什么阅读量,但实际上大大的超乎想象。所以天天讲理论看的人是越来越少,讲实际遇到的问题,并...

    AustinDatabases
  • MYSQL 5.7 VS 8 用函数你三思,三思

    最近加了几个群,里面的牛人是一个接一个,自己能不说话就不说话,主要是人家说的,看不懂呀。所以人外有人,天外有天 , 多看少说。

    AustinDatabases
  • PostgreSQL 布隆索引 与 a big bang therory

    好吧我有点标题党,其实本期要说的是 bloom 过滤器的问题,但题目为什么是这样,一般来说我们如果要给一个大表来加索引,并且这个查询还要加挺多列的时候,是蛮头疼...

    AustinDatabases
  • Python入门学习(一)

    看完了莫烦Python的视频,对于Python有了一点感觉,接下来打算把小甲鱼的视频啃完,附上学习网址:http://blog.fishc.com/catego...

    闪电gogogo
  • python:常见函数查询

    函数名: lstrip(),rstrip() 函数说明: 删除字符串开头(末尾)指定指定字符串

    用户4492257
  • Debian系统创建Sudo用户简单方法

    为保障服务器的安全性,在安装Debian系统之后,首要任务就是使用root用户登录,然后创建一个Sudo用户,目的是为了以后尽量少用root用户登录。

    IDC科技
  • python3--基础总练习题

    3、利用 python 打印前一天的本地时间,格式为‘2018-01-30’(面试题)

    py3study
  • Python面向对象4:属性、成员修饰符

    1、属性: +@property的函数,具有属性,可以直接取值。属性具有方法的写作形式,同时具有字段的访问形式,访问时不需要加()。 案例1:不加属性的调用普通...

    企鹅号小编
  • 我们来看看三种Python反转字符串方法的性能差距

    反转Python字符串的三种主要方法:“切片”,反转迭代和经典的就地反转算法。

    Python知识大全
  • 运维效率之数据迁移自动化

    起初在写overmind时就有考虑到之后的扩展,不仅仅是作为SQL自动审核执行的平台,更希望能将其打造成一个数据库自动化运维的专业系统,SQL自动审核执行作为第...

    37丫37

扫码关注云+社区

领取腾讯云代金券