使用Pandas将数据存储到SQL数据库的最佳实践

前言

在数据分析和处理过程中，我们经常需要将处理好的数据存储到数据库中以便后续使用。本文将详细介绍如何使用Pandas和SQLAlchemy将数据高效地存储到MySQL数据库中，这是数据工程中一个非常实用的技能。

准备工作

安装必要的库

首先需要安装两个Python库：

!pip3 install -U -q PyMySQL sqlalchemy

• PyMySQL：Python的MySQL客户端
• SQLAlchemy：Python的SQL工具包和对象关系映射(ORM)工具

导入必要的模块

import pandas as pd
from sqlalchemy import create_engine

数据准备阶段

下载示例数据

我们使用纽约市餐厅检查数据作为示例：

!curl 'https://data.cityofnewyork.us/api/views/43nn-pn8j/rows.csv?accessType=DOWNLOAD' -o restaurants.csv

读取和处理数据

# 读取CSV文件
df = pd.read_csv('restaurants.csv')

# 处理列名：替换空格为下划线并转为大写
cols = df.columns
cols = cols.map(lambda x: x.replace(' ', '_').upper())
df.columns = cols

# 数据类型转换
df["INSPECTION_DATE"] = pd.to_datetime(df["INSPECTION_DATE"], format="%m/%d/%Y")
df["SCORE"] = pd.to_numeric(df["SCORE"])

# 删除不需要的列
df = df.drop(['GRADE_DATE', 'RECORD_DATE', 'LOCATION_POINT1'], axis='columns')

数据库连接配置

创建数据库连接

import os
from sqlalchemy import create_engine, text

# 配置MySQL连接字符串
conn_string = 'mysql+pymysql://{user}:{password}@{host}:{port}/{db}?charset=utf8'.format(
    user     = 'student',
    password = 'dwdstudent2015',
    host     = 'db.ipeirotis.org',
    port     = 3306,
    encoding = 'utf-8',
    db = 'public'
)
engine = create_engine(conn_string)

数据库表设计

确定字段长度

在创建表之前，我们需要确定文本字段的最大长度：

# 报告文本属性的最大长度
for c in df.columns.values:
  if df.dtypes[c] == 'object':
    print(c, df[c].str.len().max())

创建唯一表名

在多用户环境下，为避免表名冲突，可以添加随机后缀：

import uuid
if "suffix" not in globals():
    suffix = str(uuid.uuid4())[:8]
print(suffix)

db_name = "public"
table_name = f"inspections_{suffix}"

创建表结构

# 删除已存在的表
drop_table_sql = f'''
DROP TABLE IF EXISTS {db_name}.{table_name};
'''

with engine.connect() as con:
    con.execute(text(drop_table_sql))

# 创建新表
create_table_sql = f'''
CREATE TABLE IF NOT EXISTS {db_name}.{table_name} (
    CAMIS CHAR(8),
    DBA VARCHAR(100),
    BUILDING VARCHAR(10),
    STREET VARCHAR(40),
    ZIPCODE CHAR(5),
    BORO VARCHAR(15),
    PHONE CHAR(12),
    CUISINE_DESCRIPTION VARCHAR(30),
    LATITUDE FLOAT,
    LONGITUDE FLOAT,
    COMMUNITY_BOARD CHAR(3),
    COUNCIL_DISTRICT CHAR(2),
    CENSUS_TRACT CHAR(6),
    BIN CHAR(7),
    BBL CHAR(10),
    NTA CHAR(4),
    INSPECTION_DATE DATETIME,
    ACTION VARCHAR(130),
    GRADE CHAR(1),
    INSPECTION_TYPE VARCHAR(60),
    VIOLATION_CODE VARCHAR(10),
    VIOLATION_DESCRIPTION VARCHAR(1000),
    CRITICAL_FLAG VARCHAR(15),
    SCORE SMALLINT
)  ENGINE=INNODB DEFAULT CHARSET=UTF8MB4;
'''

with engine.connect() as con:
    con.execute(text(create_table_sql))

数据存储

使用to_sql方法存储数据

df.to_sql(name=table_name,  # 表名
          con=engine,       # 数据库连接
          if_exists='append', # 追加模式
          index=False,      # 不写入索引列
          chunksize=1000    # 每次写入1000行
)

验证数据

# 从数据库查询数据验证
with engine.connect() as connection:
  r = pd.read_sql(text(f"SELECT * FROM public.{table_name} LIMIT 100"), con=connection)
r.head(100)

最佳实践总结

1. 数据类型处理：在将数据存储到数据库前，确保正确处理数据类型转换
2. 列名规范化：统一列名格式（如大写、下划线分隔）可以提高代码可读性
3. 表结构设计：手动设计表结构比自动推断更可靠
4. 批量写入：使用chunksize参数可以提高大数据量的写入效率
5. 多用户环境：使用随机后缀避免表名冲突
6. 字符编码：明确指定UTF8MB4以支持完整的Unicode字符集

通过以上步骤，我们可以高效地将Pandas DataFrame中的数据存储到MySQL数据库，为后续的数据分析和应用提供可靠的数据源。