数据库Excel数据导入与清洗实战指南

一、环境准备与基础操作

将 数据库 与 Excel 打通的第一步是搭建可靠的读写环境。推荐使用 Python 生态中的 pandas + SQLAlchemy 组合，它们能无缝处理常见关系型数据库（MySQL、PostgreSQL、SQLite）与 Excel 文件的交互。

1.1 安装必要依赖

pip install pandas openpyxl sqlalchemy pymysql

• pandas：读取和转换 Excel 数据
• openpyxl：处理 .xlsx 格式
• sqlalchemy：建立数据库连接和 ORM 操作
• pymysql：MySQL 驱动（如使用 PostgreSQL 则安装 psycopg2）

1.2 读取 Excel 并预览

import pandas as pd# 读取 Excel 文件，返回 DataFrame
df = pd.read_excel('sales_data.xlsx', sheet_name='Sheet1')print(df.head())          # 查看前5行
print(df.info())          # 检查每列数据类型与缺失值
print(df.describe())      # 数值列的统计摘要

此时我们已经将 Excel 数据加载到内存中，下一步就是清洗与入库。

二、实战：Excel数据入库与常见问题

2.1 建立数据库连接并写入

from sqlalchemy import create_engine# 创建数据库引擎（以 MySQL 为例）
engine = create_engine('mysql+pymysql://user:password@localhost:3306/mydb')# 将 DataFrame 写入数据库表（自动创建表）
df.to_sql(name='sales',con=engine,if_exists='replace',    # 'replace' 覆盖表，'append' 追加数据index=False,            # 不写出行索引dtype={                 # 可手动指定列类型（可选）'order_id': sqlalchemy.types.Integer(),'amount': sqlalchemy.types.Float(),'date': sqlalchemy.types.DateTime()}
)

关键点：

• if_exists='replace' 适合首次导入或全量刷新，'append' 适合增量导入。
• 通过 dtype 参数可精确控制 数据库 字段类型，避免 pandas 自动推断带来的精度问题（如日期被当作字符串）。

2.2 数据清洗：处理脏数据

从 Excel 导入时常见问题包括空值、重复行、日期格式错误、文本中包含回车换行等。以下代码演示标准的清洗流程：

# 1. 去重（基于订单ID）
df.drop_duplicates(subset='order_id', keep='first', inplace=True)# 2. 填充或删除缺失值
df.dropna(subset=['order_id', 'amount'], inplace=True)   # 关键字段缺失则删除
df['region'].fillna('未知', inplace=True)                 # 非关键列填充默认值# 3. 统一日期格式
df['date'] = pd.to_datetime(df['date'], errors='coerce') # 非法日期转为 NaT
df.dropna(subset=['date'], inplace=True)                  # 删除无法解析的日期行# 4. 文本字段清洗（去除前后空格、换行符）
df['customer_name'] = df['customer_name'].str.strip().str.replace(r'[]', '', regex=True)# 5. 数值字段检查（确保金额为正数）
df = df[df['amount'] > 0]

2.3 批量写入优化与异常处理

当 Excel 数据量超过百万行时，逐条写入速度很慢。推荐使用 pd.to_sql 的 chunksize 参数分批提交：

df.to_sql(name='sales',con=engine,if_exists='append',index=False,chunksize=5000       # 每5000行提交一次
)

同时，建议包裹异常处理，并记录失败的行日志：

from sqlalchemy.exc import IntegrityErrortry:df.to_sql(name='sales', con=engine, if_exists='append', index=False)
except IntegrityError as e:print(f"主键冲突或约束违反：{e}")# 可回退至逐行插入，跳过冲突行
else:print("数据成功导入数据库！")

通过以上实战步骤，你已经掌握了 数据库Excel 互通的完整闭环。实际生产中，建议将清洗与导入逻辑封装为定时任务（如 Airflow 或 Cron），实现 Excel 文件的自动同步，大幅提升数据管线的稳定性与效率。