### DBA Python 自动化运维练习题 以下是针对数据库管理员（DBA）使用Python进行自动化运维的一些练习题目，涵盖了常见的运维场景和技术要点。 --- #### 1. **数据库连接池管理** 设计并实现一个简单的数据库连接池类 `DatabaseConnectionPool`，该类应具备以下功能： - 支持初始化时设置最大连接数。 - 提供获取连接的方法，在达到最大连接数时阻塞等待。 - 当有空闲连接时，能够释放回连接池。 提示：可以使用 `threading.Lock()` 来处理并发访问问题[^1]。 ```python import threading class DatabaseConnectionPool: def __init__(self, max_connections): self.max_connections = max_connections self.pool = [] self.lock = threading.Lock() def acquire_connection(self): with self.lock: if len(self.pool) > 0: return self.pool.pop() elif len(self.active_connections) < self.max_connections: conn = create_new_database_connection() # 创建新的数据库连接 return conn else: raise Exception("No available connections") def release_connection(self, connection): with self.lock: self.pool.append(connection) ``` --- #### 2. **SQL 查询性能分析** 编写一个脚本，用于读取 SQL 文件中的所有查询语句，并统计每条查询的执行时间。假设存在一个函数 `execute_query(query)` 可以返回查询结果及其耗时。 提示：可以通过模拟的方式测试不同查询的执行时间。 ```python import time def analyze_sql_queries(file_path): queries = read_sql_file(file_path) results = [] for query in queries: start_time = time.time() _, execution_time = execute_query(query) end_time = time.time() total_time = end_time - start_time results.append((query, total_time)) return results def read_sql_file(path): with open(path, 'r') as file: content = file.read().split(';') return [q.strip() for q in content if q.strip()] def execute_query(query): import random sleep_time = random.uniform(0.1, 1.0) time.sleep(sleep_time) return None, sleep_time ``` --- #### 3. **备份策略自动调度** 开发一个程序，允许用户配置不同的备份频率（每日、每周、每月），并将这些任务注册到系统的定时任务计划中。提供命令行接口让用户启动或停止特定的任务。 提示：可以使用 `schedule` 库来简化定时任务的管理。 ```python import schedule import time def daily_backup(): print("Daily backup executed.") def weekly_backup(): print("Weekly backup executed.") def monthly_backup(): print("Monthly backup executed.") schedule.every().day.at("02:00").do(daily_backup) schedule.every().sunday.at("03:00").do(weekly_backup) schedule.every().month.at("04:00").do(monthly_backup) while True: schedule.run_pending() time.sleep(1) ``` [^1] --- #### 4. **日志解析与告警** 构建一个日志解析器，可以从 MySQL 错误日志或其他类型的数据库日志中提取关键信息（如错误码、时间戳、线程 ID 等字段），并将其存储到 CSV 文件中以便后续分析。如果发现严重错误，则发送邮件告警。 提示：可以使用正则表达式匹配日志格式[^1]。 ```python import re import csv import smtplib log_pattern = r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (\w+) \[(.*?)\] (.*)' def parse_log(log_file, output_csv): matches = [] critical_errors = [] with open(log_file, 'r') as f: lines = f.readlines() for line in lines: match = re.match(log_pattern, line) if match: timestamp, level, thread_id, message = match.groups() matches.append([timestamp, level, thread_id, message]) if "critical" in level.lower(): # 如果是严重错误 critical_errors.append(message) write_to_csv(output_csv, matches) send_email_alert(critical_errors) def write_to_csv(csv_file, data): headers = ['Timestamp', 'Level', 'Thread ID', 'Message'] with open(csv_file, mode='w', newline='') as file: writer = csv.writer(file) writer.writerow(headers) writer.writerows(data) def send_email_alert(errors): if not errors: return sender = "db_admin@example.com" receiver = "admin@example.com" subject = "Critical Errors Detected!" body = "\n".join(errors) try: server = smtplib.SMTP('smtp.example.com', 587) server.starttls() server.login(sender, "password") message = f"Subject: {subject}\n\n{body}" server.sendmail(sender, receiver, message) server.quit() except Exception as e: print(f"Failed to send email: {e}") ``` --- #### 5. **磁盘空间监控** 编写一个脚本，定期检查服务器上的磁盘使用情况，并在剩余空间低于某个阈值时触发告警。可以使用 `psutil` 或者 Linux 的 `df` 命令来获取磁盘状态。 提示：可以结合 `subprocess` 调用系统命令完成此操作[^4]。 ```python import subprocess def check_disk_usage(threshold=90): result = subprocess.run(['df', '-h'], stdout=subprocess.PIPE).stdout.decode('utf-8').strip() lines = result.split('\n')[1:] # 排除标题行 alerts = [] for line in lines: parts = line.split() usage_percent = int(parts[4].replace('%', '')) # 获取使用率百分比 mount_point = parts[-1] if usage_percent >= threshold: alerts.append(f"{mount_point}: Usage is at {usage_percent}%!") return alerts alerts = check_disk_usage(80) if alerts: print("\n".join(alerts)) else: print("All disks within acceptable limits.") ``` [^4] --- #### 6. **数据库表结构对比** 编写一个工具，用于比较两个数据库之间的表结构差异（列名、数据类型、约束条件等）。可以借助 SQLAlchemy 或 PyMySQL 进行元数据检索。 提示：通过反射机制获取数据库表的定义[^2]。 ```python from sqlalchemy import MetaData, Table, create_engine engine_1 = create_engine('mysql+pymysql://user:pass@host/db1') engine_2 = create_engine('mysql+pymysql://user:pass@host/db2') metadata_1 = MetaData(bind=engine_1) metadata_2 = MetaData(bind=engine_2) metadata_1.reflect() metadata_2.reflect() tables_diff = set(metadata_1.tables.keys()) ^ set(metadata_2.tables.keys()) for table_name in tables_diff: print(f"Difference found: {table_name}") # 对于相同的表，进一步比较列定义 common_tables = set(metadata_1.tables.keys()) & set(metadata_2.tables.keys()) for table_name in common_tables: columns_1 = metadata_1.tables[table_name].columns columns_2 = metadata_2.tables[table_name].columns column_names_1 = {c.name for c in columns_1} column_names_2 = {c.name for c in columns_2} diff_columns = column_names_1 ^ column_names_2 if diff_columns: print(f"Column difference in {table_name}: {diff_columns}") ``` [^2] --- ###