使用Python實現(xiàn)ELT統(tǒng)計多個服務(wù)器下所有數(shù)據(jù)表信息
更新時間：2023年07月28日 08:33:15 作者：袁袁袁袁滿
這篇文章主要介紹了使用Python實現(xiàn)ELT統(tǒng)計多個服務(wù)器下所有數(shù)據(jù)表信息,ETL，是英文Extract-Transform-Load的縮寫，用來描述將數(shù)據(jù)從來源端經(jīng)過抽取(extract)、轉(zhuǎn)換(transform)、加載(load)至目的端的過程,需要的朋友可以參考下
前言
在ETL相關(guān)工作中，將數(shù)據(jù)庫中數(shù)據(jù)的 ER結(jié)構(gòu) 整理為 Excel 可謂是一件投入產(chǎn)出比極低的、毫無技術(shù)含量的體力活
尤其是當(dāng)你有上百張表的ER結(jié)構(gòu)都要整理成 Excel 時，簡直是在燃燒生命。
大把的時間浪費在了低價值的事情上，是對時間這種資源的極大浪費。
生命短暫，更多的時間應(yīng)該投入到更有意義的事情上。
所以我寫了兩個Python腳本輕松搞定！
腳本1
效果展示
統(tǒng)計出對應(yīng)服務(wù)器，庫名所對應(yīng)表的建表時間、數(shù)據(jù)表總行數(shù)、數(shù)據(jù)表總大小、使用大小、未使用大小
代碼解析
1）導(dǎo)入所用庫
import pymssql # 操作SqlServer的庫 import openpyxl as op # 操作Excel表的庫
2）獲取服務(wù)器下所有庫名
def get_databases_name(self, cursor):
"""獲取服務(wù)器下所有庫名"""
sql = 'SELECT * FROM sys.sysdatabases'
cursor.execute(sql)
rows = cursor.fetchall() # 逐行讀取
# 存儲
databases_name = []
for list in rows:
databases_name.append(list[0])
# 移除系統(tǒng)庫
databases_name.remove("master")
databases_name.remove("model")
databases_name.remove("msdb")
databases_name.remove("tempdb")
# 移除無用庫，如果沒有則可以把try...except刪除
try:
databases_name.remove("ReportServer")
databases_name.remove("ReportServerTempDB")
except Exception as e:
print(e)
print(databases_name)
return databases_name- 提取服務(wù)器下所有數(shù)據(jù)庫名，用于后續(xù)SQL語句的拼接，利用列表的
remove方法移除不需要的數(shù)據(jù)庫
3）獲取表詳細(xì)信息并保存
def save(self, databases_name, cursor, server_name):
"""獲取表信息并保存"""
# 加入count是為了換行寫入數(shù)據(jù)
count = self.count
# databases_name:['master', 'tempdb', 'model', 'msdb', 'ReportServer', 'ReportServerTempDB', 'test']
# 取出每個庫名，用于拼接sql，獲取對應(yīng)庫名下表信息
for database in databases_name:
sql = '''
USE [%s]
SELECT a.name table_name,
a.crdate crdate,
b.rows rows,
8*b.reserved/1024 reserved,
rtrim(8*b.dpages/1024) used,
8*(b.reserved-b.dpages)/1024 unused
FROM sysobjects AS a
INNER JOIN sysindexes AS b ON a.id = b.id
WHERE ( a.type = 'u' )
AND ( b.indid IN ( 0, 1 ) )
ORDER BY a.name,b.rows DESC;
''' % database
cursor.execute(sql)
rows = cursor.fetchall() # 逐行讀取
for i in rows:
# 要寫入excel的數(shù)據(jù)
server = server_name
database_name = database
table_name = i[0]
crdate = i[1]
rows_size = i[2]
reserved = i[3]
used = i[4]
unused = i[5]
# 打印獲取到的數(shù)據(jù)
# print(server, database_name, table_name, crdate, rows_size, reserved, used, unused)
# row：count所對應(yīng)的就是行數(shù)，從第二行開始累加，colum：表示第幾列，value：表示插入的值
self.wb.cell(row=count, column=1, value=server)
self.wb.cell(row=count, column=2, value=database_name)
self.wb.cell(row=count, column=3, value=table_name)
self.wb.cell(row=count, column=4, value=crdate)
self.wb.cell(row=count, column=5, value=rows_size)
self.wb.cell(row=count, column=6, value=reserved)
self.wb.cell(row=count, column=7, value=used)
self.wb.cell(row=count, column=8, value=unused)
# count加1，進(jìn)入到下一行寫入數(shù)據(jù)
count += 1
self.count = count- 拼接SQL效果：查看表名所對應(yīng)建表時間、數(shù)據(jù)表總行數(shù)、數(shù)據(jù)表總大小、使用大小、未使用大小
4）實現(xiàn)主要邏輯
def main(self):
"""實現(xiàn)主要邏輯"""
# 1.創(chuàng)建Excel表對象，設(shè)置列名
self.ws = op.Workbook()
self.wb = self.ws.create_sheet(index=0)
self.wb.cell(row=1, column=1, value='服務(wù)器地址')
self.wb.cell(row=1, column=2, value='庫名')
self.wb.cell(row=1, column=3, value='表名')
self.wb.cell(row=1, column=4, value='建表時間')
self.wb.cell(row=1, column=5, value='數(shù)據(jù)表總行數(shù)')
self.wb.cell(row=1, column=6, value='數(shù)據(jù)表總大小(MB)')
self.wb.cell(row=1, column=7, value='使用大小(MB)')
self.wb.cell(row=1, column=8, value='未使用大小(MB)')
self.count = 2 # 定義為全局變量每次用完會更新
# 服務(wù)器列表,括號內(nèi)為：服務(wù)器名、賬號、密碼
# 如果多個服務(wù)器元組用逗號隔開
list = [('.', 'sa', 'yuan427')]
# 2.遍歷服務(wù)器列表，實現(xiàn)統(tǒng)計多個服務(wù)器
for server in list:
server_name = server[0]
user_name = server[1]
password = server[2]
conn = pymssql.connect(server_name, user_name, password)
if conn:
print("連接成功!")
cursor = conn.cursor()
# 3.獲取庫名
databases_name = self.get_databases_name(cursor)
# 4.獲取詳細(xì)信息并保存
self.save(databases_name, cursor, server_name)
# 所有服務(wù)器表插入完后保存
excel_name = "./本地數(shù)據(jù)庫統(tǒng)計.xlsx"
self.ws.save(excel_name)
# 關(guān)閉游標(biāo),關(guān)閉數(shù)據(jù)庫
cursor.close()
conn.close()1.在服務(wù)器列表循環(huán)外創(chuàng)建Excel文件
2.遍歷服務(wù)器列表，實現(xiàn)統(tǒng)計多個服務(wù)器
3.獲取當(dāng)前遍歷服務(wù)器的所有庫名
4.用庫名拼接SQL實現(xiàn)獲取數(shù)據(jù)表詳細(xì)信息
5.換行保存在Excel文件 完整代碼
需要修改的地方只有服務(wù)器地址、賬號、密碼，每一個服務(wù)器信息放一個元組中；如果有多個服務(wù)器用逗號隔開
import pymssql
import openpyxl as op
class ErTransUtils():
def get_databases_name(self, cursor):
"""獲取服務(wù)器下所有庫名"""
sql = 'SELECT * FROM sys.sysdatabases'
cursor.execute(sql)
rows = cursor.fetchall() # 逐行讀取
# 存儲
databases_name = []
for list in rows:
databases_name.append(list[0])
# 移除系統(tǒng)庫
databases_name.remove("master")
databases_name.remove("model")
databases_name.remove("msdb")
databases_name.remove("tempdb")
# 移除無用庫
try:
databases_name.remove("ReportServer")
databases_name.remove("ReportServerTempDB")
except Exception as e:
print(e)
print(databases_name)
return databases_name
def save(self, databases_name, cursor, server_name):
"""獲取表信息并保存"""
# 加入count是為了換行寫入數(shù)據(jù)
count = self.count
# databases_name:['master', 'tempdb', 'model', 'msdb', 'ReportServer', 'ReportServerTempDB', 'test'],取出每個庫名
for database in databases_name:
sql = '''
USE [%s]
SELECT a.name table_name,
a.crdate crdate,
b.rows rows,
8*b.reserved/1024 reserved,
rtrim(8*b.dpages/1024) used,
8*(b.reserved-b.dpages)/1024 unused
FROM sysobjects AS a
INNER JOIN sysindexes AS b ON a.id = b.id
WHERE ( a.type = 'u' )
AND ( b.indid IN ( 0, 1 ) )
ORDER BY a.name,b.rows DESC;
''' % database
cursor.execute(sql)
rows = cursor.fetchall() # 逐行讀取
for i in rows:
# 要寫入excel的數(shù)據(jù)
server = server_name
database_name = database
table_name = i[0]
crdate = i[1]
rows_size = i[2]
reserved = i[3]
used = i[4]
unused = i[5]
# 打印獲取到的數(shù)據(jù)
# print(server, database_name, table_name, crdate, rows_size, reserved, used, unused)
self.wb.cell(row=count, column=1, value=server)
self.wb.cell(row=count, column=2, value=database_name)
self.wb.cell(row=count, column=3, value=table_name)
self.wb.cell(row=count, column=4, value=crdate)
self.wb.cell(row=count, column=5, value=rows_size)
self.wb.cell(row=count, column=6, value=reserved)
self.wb.cell(row=count, column=7, value=used)
self.wb.cell(row=count, column=8, value=unused)
# count加1，進(jìn)入到下一行寫入數(shù)據(jù)
count += 1
self.count = count
def main(self):
"""實現(xiàn)主要邏輯"""
# 創(chuàng)建Excel表對象，設(shè)置列名
self.ws = op.Workbook()
self.wb = self.ws.create_sheet(index=0)
self.wb.cell(row=1, column=1, value='服務(wù)器地址')
self.wb.cell(row=1, column=2, value='庫名')
self.wb.cell(row=1, column=3, value='表名')
self.wb.cell(row=1, column=4, value='建表時間')
self.wb.cell(row=1, column=5, value='數(shù)據(jù)表總行數(shù)')
self.wb.cell(row=1, column=6, value='數(shù)據(jù)表總大小(MB)')
self.wb.cell(row=1, column=7, value='使用大小(MB)')
self.wb.cell(row=1, column=8, value='未使用大小(MB)')
self.count = 2
# 服務(wù)器列表,括號內(nèi)為：服務(wù)器名、賬號、密碼
# 如果多個服務(wù)器用逗號隔開
list = [('.', 'sa', 'yuan427')]
for server in list:
server_name = server[0]
user_name = server[1]
password = server[2]
conn = pymssql.connect(server_name, user_name, password)
if conn:
print("連接成功!")
cursor = conn.cursor()
databases_name = self.get_databases_name(cursor)
self.save(databases_name, cursor, server_name)
# 所有服務(wù)器表插入完后保存
excel_name = "./本地數(shù)據(jù)庫統(tǒng)計.xlsx"
self.ws.save(excel_name)
# 關(guān)閉游標(biāo),關(guān)閉數(shù)據(jù)庫
cursor.close()
conn.close()
if __name__ == '__main__':
er = ErTransUtils()
er.main()腳本2
效果展示
自動創(chuàng)建服務(wù)器文件夾，服務(wù)器文件夾下是所有庫文件夾，庫文件夾下是以表名命名的Excel文件名，文件中有表字段名稱、是否為主鍵、字段類型、字段長度、索引名稱等。
我本地的test庫
表中字段信息如下，代碼設(shè)置了Excel表格式，網(wǎng)格根據(jù)字段數(shù)量自動填充
代碼解析
由于和第一個腳本相似只講講思路：
1.獲取所有數(shù)據(jù)庫名
2.獲取庫中所有表名，把庫名和表名存放在元組內(nèi)放入列表
如：[('test', 'student', 'UTIL_IP1', 'test4', 'test5', 'test6', 'TM_AP', 'test1', 'test2', 'test3', 'UTIL_IP')]第一個是庫名其他都是表名
3.拼接獲取字段信息的SQL，把庫名、表名傳進(jìn)去，SQL能獲取到的信息如下圖（拼接的地方為上面的庫名和紅框那的表名）：
4.設(shè)置Excel內(nèi)格式：字體、加粗、居中、合并、網(wǎng)格線、行高、列寬等
5.一個Excel文件保存完畢，生成另一個表的Excel文件，只到當(dāng)前服務(wù)器下所有表統(tǒng)計完畢，才開始統(tǒng)計另一個服務(wù)器
完整代碼
需要修改的地方只有服務(wù)器地址、賬號、密碼，每一個服務(wù)器信息放一個元組中；如果有多個服務(wù)器用逗號隔開
import pymssql
import openpyxl as op
from openpyxl.styles import Font, Alignment, Side, Border
import os
class ErTransUtils():
def get_databases_name(self, cursor):
"""獲取服務(wù)器下所有庫名"""
sql = 'SELECT * FROM sys.sysdatabases'
cursor.execute(sql)
rows = cursor.fetchall() # 逐行讀取
# 存儲
databases_name = []
for list in rows:
databases_name.append(list[0])
# 移除系統(tǒng)庫和無用庫
databases_name.remove("master")
databases_name.remove("model")
databases_name.remove("msdb")
databases_name.remove("tempdb")
try:
databases_name.remove("ReportServer")
databases_name.remove("ReportServerTempDB")
except Exception as e:
print(e)
# print(databases_name)
return databases_name
def get_tables_name(self, databases_name, cursor):
"""獲取庫中所有表名，并把對應(yīng)的庫名和表名存儲在一起"""
item1 = [] # 存儲
for i in databases_name:
sql = f'SELECT * FROM [{i}].sys.tables'
cursor.execute(sql)
rows = cursor.fetchall() # 逐行讀取
item = []
for j in rows:
item.append(j[0])
item.insert(0, i)
item1.append(tuple(item))
return item1
def save(self, item1, cursor, server_name):
"""獲取表中字段信息，并寫入Excel"""
# 根據(jù)服務(wù)器名稱創(chuàng)建目錄
os.makedirs(server_name)
# databases_name:['master', 'tempdb', 'model', 'msdb', 'ReportServer', 'ReportServerTempDB', 'test']
# 取出每個庫名
for database in item1:
# 根據(jù)庫名名稱創(chuàng)建目錄
database1 = f'./{server_name}/{database[0]}'
os.makedirs(database1)
print('正在統(tǒng)計%s庫' % database[0])
for table in range(1, len(database)):
sql = '''
USE [%s]
SELECT
ColumnName=C.name,
PrimaryKey=ISNULL(IDX.PrimaryKey,N''),
Type=T.name,
Length=C.max_length,
IndexName=ISNULL(IDX.IndexName,N'')
FROM sys.columns C
INNER JOIN sys.objects O
ON C.[object_id]=O.[object_id]
AND O.type='U'
AND O.is_ms_shipped=0
INNER JOIN sys.types T
ON C.user_type_id=T.user_type_id
LEFT JOIN -- 索引及主鍵信息
(
SELECT
IDXC.[object_id],
IDXC.column_id,
Sort=CASE INDEXKEY_PROPERTY(IDXC.[object_id],IDXC.index_id,IDXC.index_column_id,'IsDescending')
WHEN 1 THEN 'DESC' WHEN 0 THEN 'ASC' ELSE '' END,
PrimaryKey=CASE WHEN IDX.is_primary_key=1 THEN N'√'ELSE N'' END,
IndexName=IDX.Name
FROM sys.indexes IDX
INNER JOIN sys.index_columns IDXC
ON IDX.[object_id]=IDXC.[object_id]
AND IDX.index_id=IDXC.index_id
INNER JOIN -- 對于一個列包含多個索引的情況,只顯示第1個索引信息
(
SELECT [object_id], Column_id, index_id=MIN(index_id)
FROM sys.index_columns
GROUP BY [object_id], Column_id
) IDXCUQ
ON IDXC.[object_id]=IDXCUQ.[object_id]
AND IDXC.Column_id=IDXCUQ.Column_id
AND IDXC.index_id=IDXCUQ.index_id
) IDX
ON C.[object_id]=IDX.[object_id]
AND C.column_id=IDX.column_id
WHERE O.name=N'%s' -- 修改表名
ORDER BY O.name,C.column_id
''' % (database[0], database[table])
# 執(zhí)行sql語句
try:
cursor.execute(sql)
rows = cursor.fetchall() # 逐行讀取
except Exception as e:
print(e)
# 存儲
lists = []
for list in rows:
lists.append(list)
# 獲取字段的行數(shù),+2表示Excel的行數(shù)
excel_line = len(lists) + 2
# 加入count是為了換行寫入數(shù)據(jù)
count = 3
wb = op.Workbook()
ws = wb.create_sheet(index=0)
table_name = f'所統(tǒng)計數(shù)據(jù)表:({database[table]})'
ws.cell(row=1, column=1, value=table_name)
ws.cell(row=2, column=1, value='中文名稱')
ws.cell(row=2, column=2, value='數(shù)據(jù)庫字段名稱')
ws.cell(row=2, column=3, value='是否為主鍵')
ws.cell(row=2, column=4, value='字段類型')
ws.cell(row=2, column=5, value='字段長度')
ws.cell(row=2, column=6, value='是否多值')
ws.cell(row=2, column=7, value='索引名稱')
ws.cell(row=2, column=8, value='數(shù)據(jù)樣本')
ws.cell(row=2, column=9, value='應(yīng)用說明')
# 整體格式
row_range = ws[1:excel_line]
for row in row_range:
for cell in row:
cell.font = Font(name="等線", size=12, bold=False, italic=False,
color="00000000") # name=字體名稱,size=字體大小,bold=是否加粗,color=字體顏色
cell.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True) # 字體上下左右居中
side1 = Side(style="thin", color="00000000")
side2 = Side(style="thin", color="00000000")
cell.border = Border(left=side1, right=side1, top=side2, bottom=side2) # 邊框
# 設(shè)置列寬
ws.column_dimensions['A'].width = 14
ws.column_dimensions['B'].width = 16
ws.column_dimensions['C'].width = 14
ws.column_dimensions['D'].width = 14
ws.column_dimensions['E'].width = 14
ws.column_dimensions['F'].width = 14
ws.column_dimensions['G'].width = 20
ws.column_dimensions['H'].width = 22
ws.column_dimensions['I'].width = 29
# 單獨設(shè)置應(yīng)用說明列
column = f'I3:I{excel_line}'
font1 = ws[column]
for a in font1:
for a1 in a:
a1.alignment = Alignment(horizontal="left", vertical="justify", wrap_text=True)
# 設(shè)置第一行格式
ws.row_dimensions[1].height = 46.5 # 行高
ws.merge_cells('A1:I1') # 合并單元格
cell = ws["A1"]
cell.font = Font(name="等線", size=16, bold=True, italic=False,
color="00000000") # name=字體名稱,size=字體大小,bold=是否加粗,color=字體顏色
# 設(shè)置第二行格式
ws.row_dimensions[2].height = 24 # 行高
font2 = ws['A2:I2']
for b in font2:
for b1 in b:
b1.font = Font(name="等線", size=12, bold=True, italic=False,
color="00000000") # name=字體名稱,size=字體大小,bold=是否加粗,color=字體顏色
for i in lists:
# 要寫入excel的數(shù)據(jù)
field_name = i[0]
key_name = i[1]
field_type = i[2]
field_length = i[3]
index_name = i[4]
# 將數(shù)據(jù)寫入到下一行
ws.cell(row=count, column=2, value=field_name)
ws.cell(row=count, column=3, value=key_name)
ws.cell(row=count, column=4, value=field_type)
ws.cell(row=count, column=5, value=field_length)
ws.cell(row=count, column=7, value=index_name)
# count加1，進(jìn)入到下一行寫入數(shù)據(jù)
count += 1
excel_name = f'./{server_name}/{database[0]}/{database[table]}.xlsx'
wb.save(excel_name)
print('%s庫統(tǒng)計完成' % database[0])
def main(self):
# 服務(wù)器列表
list = [('.', 'sa', 'yuan427')]
for server in list:
server_name = server[0]
user_name = server[1]
password = server[2]
conn = pymssql.connect(server_name, user_name, password)
if conn:
print("連接成功!")
cursor = conn.cursor()
databases_name = self.get_databases_name(cursor)
item1 = self.get_tables_name(databases_name, cursor)
print(item1)
self.save(item1, cursor, server_name)
# 關(guān)閉游標(biāo),關(guān)閉數(shù)據(jù)庫
cursor.close()
conn.close()
if __name__ == '__main__':
er = ErTransUtils()
er.main()到此這篇關(guān)于使用Python實現(xiàn)ELT統(tǒng)計多個服務(wù)器下所有數(shù)據(jù)表信息的文章就介紹到這了,更多相關(guān)Python實現(xiàn)ELT統(tǒng)計內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
Tensorflow2.10實現(xiàn)圖像分割任務(wù)示例詳解
這篇文章主要為大家介紹了Tensorflow2.10實現(xiàn)圖像分割任務(wù)示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-01-01
Python numpy生成矩陣、串聯(lián)矩陣代碼分享
這篇文章主要介紹了Python numpy生成矩陣、串聯(lián)矩陣代碼分享，具有一定參考價值，需要的朋友可以了解下。2017-12-12
這篇文章主要為大家詳細(xì)介紹了python實現(xiàn)桌面托盤氣泡提示，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-07-07
Python網(wǎng)絡(luò)編程之使用TCP方式傳輸文件操作示例
這篇文章主要介紹了Python網(wǎng)絡(luò)編程之使用TCP方式傳輸文件操作,結(jié)合實例形式分析了使用socket模塊進(jìn)行tcp協(xié)議下文件傳輸?shù)脑硪约胺?wù)器端、客戶端相關(guān)實現(xiàn)技巧,需要的朋友可以參考下2019-11-11
Pycharm 使用 Pipenv 新建的虛擬環(huán)境(圖文詳解)
pipenv 是 Pipfile 主要倡導(dǎo)者、requests 作者 Kenneth Reitz 寫的一個命令行工具，主要包含了Pipfile、pip、click、requests和virtualenv。這篇文章主要介紹了Pycharm 使用 Pipenv 新建的虛擬環(huán)境的問題,需要的朋友可以參考下2020-04-04
解決Tkinter中button按鈕未按卻主動執(zhí)行command函數(shù)的問題
這篇文章主要介紹了解決Tkinter中button按鈕未按卻主動執(zhí)行command函數(shù)的問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05
python使用Tkinter顯示網(wǎng)絡(luò)圖片的方法
這篇文章主要介紹了python使用Tkinter顯示網(wǎng)絡(luò)圖片的方法,涉及Python操作圖片的相關(guān)技巧,需要的朋友可以參考下2015-04-04
Python Pandas中合并數(shù)據(jù)的5個函數(shù)使用詳解
數(shù)據(jù)合并是數(shù)據(jù)處理過程中的必經(jīng)環(huán)節(jié)，pandas作為數(shù)據(jù)分析的利器，提供了五種常用的數(shù)據(jù)合并方式，讓我們看看如何使用這些方法吧！2022-05-05
python實現(xiàn)掃描局域網(wǎng)指定網(wǎng)段ip的方法
這篇文章主要介紹了python實現(xiàn)掃描局域網(wǎng)指定網(wǎng)段ip的方法，非常不錯，具有一定的參考借鑒價值,需要的朋友可以參考下2019-04-04
今天小編就為大家分享一篇關(guān)于Python字符串逆序輸出的實例講解，小編覺得內(nèi)容挺不錯的，現(xiàn)在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧2019-02-02