AI 助理
文档备案控制台
开发者社区 数据库 文章 正文

两个Python脚本轻松解决ETL工作：统计多个服务器下所有数据表信息

2024-05-17 199
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 两个Python脚本轻松解决ETL工作：统计多个服务器下所有数据表信息

正文

服务器列表,括号内为：服务器名、账号、密码

如果多个服务器元组用逗号隔开

list = [(‘.’, ‘sa’, ‘yuan427’)]

2.遍历服务器列表，实现统计多个服务器

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.获取详细信息并保存

self.save(databases_name, cursor, server_name)

所有服务器表插入完后保存

excel_name = “./本地数据库统计.xlsx”
self.ws.save(excel_name)

关闭游标,关闭数据库

cursor.close()
conn.close()
  • 1.在服务器列表循环外创建Excel文件
  • 2.遍历服务器列表，实现统计多个服务器
  • 3.获取当前遍历服务器的所有库名
  • 4.用库名拼接SQL实现获取数据表详细信息
  • 5.换行保存在Excel文件

完整代码

需要修改的地方只有服务器地址、账号、密码，每一个服务器信息放一个元组中；如果有多个服务器用逗号隔开

import pymssql
import openpyxl as op
class ErTransUtils():
def get_databases_name(self, cursor):
“”“获取服务器下所有库名”“”
sql = ‘SELECT * FROM sys.sysdatabases’
cursor.execute(sql)
rows = cursor.fetchall() # 逐行读取

存储

databases_name = []
for list in rows:
databases_name.append(list[0])

移除系统库

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是为了换行写入数据

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的数据

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]

打印获取到的数据

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，进入到下一行写入数据

count += 1
self.count = count
def main(self):
“”“实现主要逻辑”“”

创建Excel表对象，设置列名

self.ws = op.Workbook()
self.wb = self.ws.create_sheet(index=0)
self.wb.cell(row=1, column=1, value=‘服务器地址’)
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=‘数据表总行数’)
self.wb.cell(row=1, column=6, value=‘数据表总大小(MB)’)
self.wb.cell(row=1, column=7, value=‘使用大小(MB)’)
self.wb.cell(row=1, column=8, value=‘未使用大小(MB)’)
self.count = 2

服务器列表,括号内为：服务器名、账号、密码

如果多个服务器用逗号隔开

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)

所有服务器表插入完后保存

excel_name = “./本地数据库统计.xlsx”
self.ws.save(excel_name)

关闭游标,关闭数据库

cursor.close()
conn.close()
if name == ‘main’:
er = ErTransUtils()
er.main()

脚本2

=====================================================================

效果展示

自动创建服务器文件夹，服务器文件夹下是所有库文件夹，库文件夹下是以表名命名的Excel文件名，文件中有表字段名称、是否为主键、字段类型、字段长度、索引名称等。

  • 我本地的test库

  • 表中字段信息如下，代码设置了Excel表格式，网格根据字段数量自动填充

代码解析

由于和第一个脚本相似只讲讲思路：

  • 1.获取所有数据库名
  • 2.获取库中所有表名，把库名和表名存放在元组内放入列表，如：[('test', 'student', 'UTIL_IP1', 'test4', 'test5', 'test6', 'TM_AP', 'test1', 'test2', 'test3', 'UTIL_IP')]，第一个是库名其他都是表名
  • 3.拼接获取字段信息的SQL，把库名、表名传进去，SQL能获取到的信息如下图（拼接的地方为上面的库名和红框那的表名）：

  • 4.设置Excel内格式：字体、加粗、居中、合并、网格线、行高、列宽等
  • 5.一个Excel文件保存完毕，生成另一个表的Excel文件，只到当前服务器下所有表统计完毕，才开始统计另一个服务器

完整代码

需要修改的地方只有服务器地址、账号、密码，每一个服务器信息放一个元组中；如果有多个服务器用逗号隔开

import pymssql
import openpyxl as op
from openpyxl.styles import Font, Alignment, Side, Border
import os
class ErTransUtils():
def get_databases_name(self, cursor):
“”“获取服务器下所有库名”“”
sql = ‘SELECT * FROM sys.sysdatabases’
cursor.execute(sql)
rows = cursor.fetchall() # 逐行读取

存储

databases_name = []
for list in rows:
databases_name.append(list[0])

移除系统库和无用库

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):
“”“获取库中所有表名，并把对应的库名和表名存储在一起”“”
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”“”

根据服务器名称创建目录

os.makedirs(server_name)
databases_name:[‘master’, ‘tempdb’, ‘model’, ‘msdb’, ‘ReportServer’, ‘ReportServerTempDB’, ‘test’]

取出每个库名

for database in item1:

根据库名名称创建目录

database1 = f’./{server_name}/{database[0]}’
os.makedirs(database1)
print(‘正在统计%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])

执行sql语句

try:
cursor.execute(sql)
rows = cursor.fetchall() # 逐行读取
except Exception as e:
print(e)

存储

lists = []
for list in rows:
lists.append(list)

获取字段的行数,+2表示Excel的行数

excel_line = len(lists) + 2

加入count是为了换行写入数据

count = 3
wb = op.Workbook()
ws = wb.create_sheet(index=0)
table_name = f’所统计数据表:({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=‘数据库字段名称’)
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=‘数据样本’)
ws.cell(row=2, column=9, value=‘应用说明’)

整体格式

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) # 边框

设置列宽

感谢每一个认真阅读我文章的人，看着粉丝一路的上涨和关注，礼尚往来总是要有的：

①　2000多本Python电子书（主流和经典的书籍应该都有了）

②　Python标准库资料（最全中文版）

③　项目源码（四五十个有趣且经典的练手项目及源码）

④　Python基础入门、爬虫、web开发、大数据分析方面的视频（适合小白学习）

⑤ Python学习路线图（告别不入流的学习）


文章标签：
SQL
Python
数据库
存储
数据安全/隐私保护
关键词：
Python服务器
python云服务器 ECS
Python脚本
云服务器 ECS python
Python统计
老板这功能得加钱
目录
相关文章
pickstar-33775
|
10月前
|
人工智能 JavaScript API
零基础构建MCP服务器：TypeScript/Python双语言实战指南
作为一名深耕技术领域多年的博主摘星，我深刻感受到了MCP（Model Context Protocol）协议在AI生态系统中的革命性意义。MCP作为Anthropic推出的开放标准，正在重新定义AI应用与外部系统的交互方式，它不仅解决了传统API集成的复杂性问题，更为开发者提供了一个统一、安全、高效的连接框架。在过去几个月的实践中，我发现许多开发者对MCP的概念理解透彻，但在实际动手构建MCP服务器时却遇到了各种技术壁垒。从环境配置的细节问题到SDK API的深度理解，从第一个Hello World程序的调试到生产环境的部署优化，每一个环节都可能成为初学者的绊脚石。因此，我决定撰写这篇全面的实
pickstar-33775
2272 67
零基础构建MCP服务器：TypeScript/Python双语言实战指南
Star时光
|
11月前
|
Python
掌握Python装饰器：轻松统计函数执行时间
掌握Python装饰器：轻松统计函数执行时间
Star时光
583 76
站大爷
|
9月前
|
人工智能 自然语言处理 安全
Python构建MCP服务器：从工具封装到AI集成的全流程实践
MCP协议为AI提供标准化工具调用接口，助力模型高效操作现实世界。
站大爷
1532 1
站大爷
|
10月前
|
安全 Linux 网络安全
Python极速搭建局域网文件共享服务器：一行命令实现HTTPS安全传输
本文介绍如何利用Python的http.server模块，通过一行命令快速搭建支持HTTPS的安全文件下载服务器，无需第三方工具，3分钟部署，保障局域网文件共享的隐私与安全。
站大爷
2519 0
异步社区
|
数据挖掘 索引 Python
《从Excel到Python——数据分析进阶指南》一第1章 生成数据表
异步社区
4502 0
九月天空
|
8月前
|
数据采集 机器学习/深度学习 人工智能
Python：现代编程的首选语言
Python：现代编程的首选语言
九月天空
1277 102
九月天空
|
8月前
|
数据采集 机器学习/深度学习 算法框架/工具
Python：现代编程的瑞士军刀
Python：现代编程的瑞士军刀
九月天空
461 104
九月天空
|
8月前
|
人工智能 自然语言处理 算法框架/工具
Python：现代编程的首选语言
Python：现代编程的首选语言
九月天空
356 103
九月天空
|
8月前
|
机器学习/深度学习 人工智能 数据挖掘
Python：现代编程的首选语言
Python：现代编程的首选语言
九月天空
370 82

热门文章

最新文章

  • 1
    ECS 按量付费VPC实例停机不收费FAQ
  • 2
    阿里云香港云服务器ECS、轻量应用服务器和云虚拟主机
  • 3
    阿里云上搭建webRTC 服务器——Licode
  • 4
    Linux服务器中了病毒后的清理方法
  • 5
    阿里云快照怎么收费？云服务器快照收费标准了解一下
  • 6
    阿里云服务器ecs配置之安装mysql
  • 7
    VPC DHCP类型的ECS修改DNS
  • 8
    mysql5.7 solo web服务器
  • 9
    阿里云服务器2核4G、4核8G、8核16G配置活动价格及实例规格选择参考（2024更新）
  • 10
    Linux 服务器建站新手教程不需要敲一行命令【小白篇】
  • 1
    字典在Python中的应用与实例
    489
  • 2
    元组（Tuple）在Python编程中的应用与实例
    1093
  • 3
    Python函数：深入理解与应用
    222
  • 4
    Python中的try-except异常处理语句
    1162
  • 5
    可移植性在Python中的优势
    628
  • 6
    Python与C的联姻：探索Python的可移植性与C的底层力量
    482
  • 7
    如何选择适合的Python并行计算库
    411
  • 8
    如何优化Python代码以提高执行效率
    180
  • 9
    Python编程的深入探索与实用案例
    192
  • 10
    聊聊python多线程与多进程
    259

    • 相关课程

    更多
  • Python Web开发基础
  • Python爬虫实战
  • Python常用数据科学库
  • Python网络爬虫实战
  • Python完全自学手册图文教程
  • Python基础快速入门实战教程

    • 相关电子书

    更多
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:
  • 双剑合璧-Python和大数据计算平台的结合

    • 推荐镜像

    更多
  • python-release
    • 下一篇
    阿里云Agent安全中心领跑IDC 2026智能体威胁检测技术评估