python+mysql+线程池

import mysql.connector from concurrent.futures import ThreadPoolExecutor from typing import List, Dict import time class MySQLDataExtractor: def __init__(self, host: str, user: str, password: str, database: str): self.connection_config = { 'host': host, 'user': user, 'password': password, 'database': database } def get_connection(self): """创建数据库连接""" return mysql.connector.connect(**self.connection_config) def fetch_chunk(self, query: str, params: tuple = None) -> pd.DataFrame: """执行单个查询并返回结果""" try: conn = self.get_connection() df = pd.read_sql(query, conn, params=params) return df except Exception as e: print(f"Error fetching chunk: {e}") return pd.DataFrame() finally: conn.close() def parallel_fetch(self, base_query: str, chunk_size: int = 10000, max_workers: int = 4) -> pd.DataFrame: """并行获取数据""" # 首先获取总记录数 count_query = f"SELECT COUNT(*) as count FROM ({base_query}) as t" with self.get_connection() as conn: total_count = pd.read_sql(count_query, conn).iloc[0]['count'] # 计算需要的分块数 chunks = [] for offset in range(0, total_count, chunk_size): chunk_query = f"{base_query} LIMIT {chunk_size} OFFSET {offset}" chunks.append(chunk_query) # 使用线程池并行处理 start_time = time.time() results = [] with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(self.fetch_chunk, query) for query in chunks] for future in futures: try: result = future.result() results.append(result) except Exception as e: print(f"Error in thread: {e}") end_time = time.time() print(f"Total time taken: {end_time - start_time:.2f} seconds") # 合并所有结果 return pd.concat(results, ignore_index=True) def main(): # 配置数据库连接 extractor = MySQLDataExtractor( host="your_host", user="your_username", password="your_password", database="your_database" ) # 示例查询 query = """ SELECT * FROM your_table WHERE some_condition = 1 """ # 使用4个线程并行提取数据，每个块10000条记录 df = extractor.parallel_fetch( base_query=query, chunk_size=10000, max_workers=4 ) print(f"Total records fetched: {len(df)}") print(df.head()) if __name__ == "__main__": main()