使用Pandas将数据高效插入MySQL数据库 - ipeirotis/dealing_with_data项目实践

前言

在数据分析和处理过程中，我们经常需要将处理好的数据存储到数据库中以便后续使用。Python的Pandas库提供了非常便捷的方法来实现这一需求。本文将通过ipeirotis/dealing_with_data项目中的一个实际案例，详细介绍如何使用Pandas将数据高效地插入MySQL数据库。

准备工作

在开始之前，我们需要安装必要的Python库：

!pip install -U -q PyMySQL sqlalchemy

这两个库分别用于：

• PyMySQL：Python连接MySQL数据库的驱动
• SQLAlchemy：Python SQL工具包和ORM框架

获取数据源

本案例使用纽约市Citibike共享单车系统的公开API作为数据源。我们主要关注两个API端点：

# 站点基本信息（相对稳定）
url_stations = "https://gbfs.citibikenyc.com/gbfs/en/station_information.json"
# 站点实时状态（动态变化）
url_status = "https://gbfs.citibikenyc.com/gbfs/en/station_status.json"

首先获取站点基本信息：

import requests
results = requests.get(url_stations).json()
data = results["data"]["stations"]

数据处理与DataFrame创建

将获取的JSON数据转换为Pandas DataFrame：

import pandas as pd
df = pd.DataFrame(data)

查看数据前5行：

df.head(5)

由于原始数据中包含一些不适合直接存入数据库的字段（如包含多个值的列表），我们需要进行清理：

df.drop(
    ["rental_methods", "eightd_station_services", "rental_uris"],
    axis="columns",
    inplace=True,
)

数据库连接配置

使用SQLAlchemy创建数据库连接引擎：

import sqlalchemy
from sqlalchemy import create_engine, text

conn_string = "mysql+pymysql://{user}:{password}@{host}/".format(
    host="db.ipeirotis.org", user="student", password="dwdstudent2015"
)

engine = create_engine(conn_string)

数据库和表创建

创建数据库

db_name = "public"
create_db_query = (
    f"CREATE DATABASE IF NOT EXISTS {db_name} DEFAULT CHARACTER SET 'utf8'"
)

with engine.connect() as con:
    con.execute(text(create_db_query))
    con.execute(text(f"USE {db_name}"))

创建表结构

为了避免多人协作时的表名冲突，我们为表名添加随机后缀：

import uuid
suffix = str(uuid.uuid4())[:8]
table_name = f"Stations_{suffix}"

定义表结构：

create_table_query = f"""CREATE TABLE IF NOT EXISTS {db_name}.{table_name}
                        (legacy_id int,
                        station_id varchar(100),
                        region_id int,
                        external_id varchar(50),
                        lat float,
                        lon float,
                        short_name varchar(10),
                        name varchar(250),
                        station_type varchar(10),
                        capacity int,
                        electric_bike_surcharge_waiver bool,
                        eightd_has_key_dispenser bool,
                        has_kiosk bool,
                        PRIMARY KEY(station_id)
                        )"""

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

数据插入

使用Pandas的to_sql方法将DataFrame数据插入数据库：

df.to_sql(
    name=table_name,
    schema=db_name,
    con=engine,
    if_exists="append",
    index=False
)

数据查询

从数据库中读取数据同样简单：

query = text(f"SELECT * FROM {db_name}.{table_name}")

with engine.connect() as con:
    df2 = pd.read_sql(query, con=con)
df2.head(5)

数据导出

可以将查询结果导出为Excel或CSV格式：

df2.to_excel("citibike.xlsx")
df2.to_csv("citibike.csv")

清理工作

使用完毕后删除临时表：

drop_table_query = f"DROP TABLE IF EXISTS {db_name}.{table_name}"

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

扩展练习

尝试获取并存储站点的实时状态数据，并建立与站点基本信息表的外键关系：

url_status = 'https://gbfs.citibikenyc.com/gbfs/en/station_status.json'
# 获取数据并处理
# 创建状态表
# 建立外键关系

总结

通过本案例，我们学习了：

1. 如何使用Pandas处理API获取的JSON数据
2. 如何配置MySQL数据库连接
3. 如何定义数据库表结构
4. 使用Pandas的to_sql方法高效插入数据
5. 使用read_sql方法从数据库读取数据
6. 数据导出和清理的最佳实践

这种方法特别适合数据分析师和数据工程师，可以大大简化数据存储和读取的流程，提高工作效率。