<think>我们通常使用以下步骤将Excel文件导入MySQL： 1. 将Excel文件另存为CSV格式（因为CSV更容易处理） 2. 使用编程语言（如Python）读取CSV文件并处理数据 3. 连接MySQL数据库并创建表（如果不存在） 4. 将数据批量插入到MySQL表中 这里提供一个使用Python的完整解决方案，需要安装的库：`pandas` 和 `mysql-connector-python` 安装命令：`pip install pandas mysql-connector-python` 步骤： 1. 读取Excel文件（支持xls和xlsx格式） 2. 将数据转换为适合MySQL的格式（处理空值、日期等） 3. 建立数据库连接 4. 创建表（根据DataFrame的列自动生成） 5. 将数据插入到MySQL表中 注意：在实际应用中，我们可能需要考虑数据类型映射、数据清洗、异常处理等。 以下是一个完整的示例代码： </think> ### Excel 文件导入 MySQL 的完整解决方案 将 Excel 数据导入 MySQL 数据库是常见的数据处理任务，以下是 Python + Pandas 的完整解决方案： #### 核心代码实现 ```python import pandas as pd from sqlalchemy import create_engine, exc import logging # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def excel_to_mysql(excel_path, db_config, table_name, sheet_name=0, if_exists='replace'): """ 将 Excel 文件导入 MySQL 数据库 参数: excel_path: Excel 文件路径 db_config: 数据库配置字典 table_name: 目标表名 sheet_name: Excel 工作表名或索引 if_exists: 表存在时的处理方式 ('replace', 'append', 'fail') """ try: # 1. 读取 Excel 文件 logging.info(f"开始读取 Excel 文件: {excel_path}") df = pd.read_excel(excel_path, sheet_name=sheet_name) logging.info(f"成功读取数据: {len(df)} 行, {len(df.columns)} 列") # 2. 数据预处理 df = preprocess_data(df) # 3. 创建数据库连接 engine = create_engine( f"mysql+mysqlconnector://{db_config['user']}:{db_config['password']}" f"@{db_config['host']}:{db_config['port']}/{db_config['database']}" ) # 4. 导入数据到 MySQL logging.info(f"开始导入数据到表: {table_name}") df.to_sql( name=table_name, con=engine, index=False, # 不保存索引列 if_exists=if_exists, chunksize=1000 # 分块导入 ) logging.info("数据导入成功!") return True except exc.SQLAlchemyError as e: logging.error(f"数据库错误: {str(e)}") return False except Exception as e: logging.error(f"处理失败: {str(e)}") return False def preprocess_data(df): """数据预处理函数""" # 1. 处理空值 df = df.fillna(pd.NA) # 统一空值表示 # 2. 处理日期时间列 date_cols = df.select_dtypes(include=['datetime']).columns for col in date_cols: df[col] = df[col].dt.strftime('%Y-%m-%d %H:%M:%S') # 3. 清理列名 (移除特殊字符) df.columns = [col.replace(' ', '_').replace('.', '').lower() for col in df.columns] # 4. 截断超长文本 (MySQL TEXT 类型最多 65,535 字符) str_cols = df.select_dtypes(include=['object']).columns for col in str_cols: df[col] = df[col].apply(lambda x: str(x)[:65500] if isinstance(x, str) else x) return df # 使用示例 if __name__ == "__main__": # 数据库配置 db_config = { 'host': '10.137.26.51', 'port': 3306, 'user': 'cwx', 'password': 'x79157915!', 'database': 'app_db' } # 执行导入 success = excel_to_mysql( excel_path='data/sales_report.xlsx', db_config=db_config, table_name='sales_data', sheet_name='Sheet1', if_exists='replace' # 覆盖现有表 ) if success: print("Excel 数据成功导入 MySQL!") else: print("导入失败，请检查日志") ``` #### 代码详解 1. **Excel 读取**： - 使用 `pd.read_excel()` 读取 Excel 文件 - 支持多工作表选择（通过 `sheet_name` 参数） - 自动处理 Excel 的各种格式（xls/xlsx） 2. **数据预处理**： - 空值处理：统一转换为 Pandas 的 `pd.NA` - 日期格式化：转换为 MySQL 兼容的日期字符串 - 列名清理：移除特殊字符，转为小写 - 文本截断：防止超长字符串导致导入失败 3. **数据库连接**： - 使用 SQLAlchemy 创建连接引擎 - 支持多种数据库驱动（这里使用 mysqlconnector） - 自动管理连接池 4. **数据导入**： - `to_sql()` 方法实现批量导入 - `chunksize=1000` 分块提交，避免内存溢出 - `if_exists` 参数控制表存在时的行为 #### 高级功能扩展 ```python # 添加到 excel_to_mysql 函数中 def excel_to_mysql(...): ... # 添加数据类型映射 dtype_mapping = { 'int64': 'BIGINT', 'float64': 'DOUBLE', 'datetime64': 'DATETIME', 'object': 'TEXT' } # 自定义数据类型 custom_types = { 'price': 'DECIMAL(10,2)', 'product_code': 'VARCHAR(20)' } # 执行导入（添加 dtype 参数） df.to_sql( ... dtype={**{col: dtype_mapping[str(df[col].dtype)] for col in df.columns}, **custom_types} ) ... ``` #### 替代方案：MySQL LOAD DATA INFILE 对于大型 Excel 文件（>100MB），推荐转换为 CSV 后使用 MySQL 原生导入： ```sql -- 1. 转换 Excel 为 CSV # 使用 pandas: df.to_csv('data.csv', index=False) -- 2. MySQL 导入命令 LOAD DATA LOCAL INFILE '/path/to/data.csv' INTO TABLE sales_data FIELDS TERMINATED BY ',' ENCLOSED BY '"' LINES TERMINATED BY '\n' IGNORE 1 ROWS; -- 跳过标题行 ``` #### 错误处理增强 ```python # 添加到 excel_to_mysql 函数中 try: ... except pd.errors.EmptyDataError: logging.error("Excel 文件为空或格式错误") except pd.errors.ParserError: logging.error("Excel 解析错误，请检查文件格式") except exc.OperationalError as e: if "Access denied" in str(e): logging.error("数据库访问被拒绝，请检查凭据") elif "Unknown database" in str(e): logging.error("数据库不存在，请先创建数据库") else: logging.error(f"数据库连接错误: {str(e)}") ``` #### 性能优化建议 1. **批量处理**： ```python # 使用迭代器分块读取大型 Excel chunk_size = 10000 with pd.read_excel(excel_path, chunksize=chunk_size) as reader: for chunk in reader: chunk.to_sql(...) ``` 2. **禁用索引**： ```python # 导入前禁用索引，导入后重建 with engine.connect() as conn: conn.execute(f"ALTER TABLE {table_name} DISABLE KEYS") df.to_sql(...) conn.execute(f"ALTER TABLE {table_name} ENABLE KEYS") ``` 3. **并行处理**： ```python from concurrent.futures import ThreadPoolExecutor def import_chunk(chunk): chunk.to_sql(...) with ThreadPoolExecutor(max_workers=4) as executor: executor.map(import_chunk, pd.read_excel(..., chunksize=5000)) ``` ### 完整工作流程 1. **准备阶段**： - 检查 Excel 文件格式 - 确认数据库连接权限 - 创建目标数据库表结构 2. **执行导入**： - 读取 Excel → 数据清洗 → 分块导入 - 实时监控进度和错误 3. **验证结果**： ```sql -- 检查行数 SELECT COUNT(*) FROM sales_data; -- 抽样检查数据 SELECT * FROM sales_data LIMIT 5; -- 检查数据完整性 SELECT column_name, COUNT(*) null_count FROM sales_data GROUP BY column_name; ```