快捷導(dǎo)航

Pandas使用SQLite3實(shí)戰(zhàn)

更新時(shí)間：2025年04月02日 10:09:04 作者：老哥不老

本文主要介紹了Pandas使用SQLite3實(shí)戰(zhàn),文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

1 環(huán)境準(zhǔn)備

確保已安裝 pandas 和 sqlite3（前者需單獨(dú)安裝，后者是 Python 內(nèi)置）：

pip install pandas

2 從 SQLite3 讀取數(shù)據(jù)到 DataFrame

基礎(chǔ)用法：讀取整個(gè)表

import pandas as pd
import sqlite3

# 連接到數(shù)據(jù)庫(kù)
conn = sqlite3.connect('test.db')

# 讀取 users 表到 DataFrame
df = pd.read_sql('SELECT * FROM users', conn)
print(df.head())  # 查看前5行數(shù)據(jù)

# 關(guān)閉連接
conn.close()

高級(jí)用法：篩選和聚合

query = '''
    SELECT 
        name, 
        AVG(age) as avg_age   -- 計(jì)算平均年齡
    FROM users 
    WHERE age > 20 
    GROUP BY name
'''
df = pd.read_sql(query, conn)
print(df)

3 將 DataFrame 寫(xiě)入 SQLite3

基本寫(xiě)入（全量覆蓋）

# 創(chuàng)建一個(gè)示例 DataFrame
data = {
    'name': ['David', 'Eve'],
    'age': [28, 32],
    'email': ['david@test.com', 'eve@test.com']
}
df = pd.DataFrame(data)

# 寫(xiě)入到 users 表（全量覆蓋）
df.to_sql(
    name='users',     # 表名
    con=conn,         # 數(shù)據(jù)庫(kù)連接
    if_exists='replace',  # 如果表存在，直接替換（慎用?。?
    index=False       # 不保存 DataFrame 的索引列
)
conn.commit()

追加數(shù)據(jù)（增量寫(xiě)入）

df.to_sql(
    name='users',
    con=conn,
    if_exists='append',  # 追加到現(xiàn)有表
    index=False
)
conn.commit()

4 實(shí)戰(zhàn)場(chǎng)景：數(shù)據(jù)清洗 + 入庫(kù)

假設(shè)有一個(gè) CSV 文件 dirty_data.csv，需要清洗后存入 SQLite3：

id,name,age,email
1, Alice,30,alice@example.com
2, Bob , invalid, bob@example.com  # 錯(cuò)誤年齡
3, Charlie,35,missing_email

步驟 1：用 Pandas 清洗數(shù)據(jù)

# 讀取 CSV
df = pd.read_csv('dirty_data.csv')

# 清洗操作
df['age'] = pd.to_numeric(df['age'], errors='coerce')  # 無(wú)效年齡轉(zhuǎn)為 NaN
df = df.dropna(subset=['age'])                        # 刪除年齡無(wú)效的行
df['email'] = df['email'].fillna('unknown')            # 填充缺失郵箱
df['name'] = df['name'].str.strip()                   # 去除名字前后空格

print(df)

步驟 2：寫(xiě)入數(shù)據(jù)庫(kù)

with sqlite3.connect('test.db') as conn:
    # 寫(xiě)入新表 cleaned_users
    df.to_sql('cleaned_users', conn, index=False, if_exists='replace')
    
    # 驗(yàn)證寫(xiě)入結(jié)果
    df_check = pd.read_sql('SELECT * FROM cleaned_users', conn)
    print(df_check)

5 性能優(yōu)化：分塊寫(xiě)入大數(shù)據(jù)

處理超大型數(shù)據(jù)時(shí)（如 10 萬(wàn)行），避免一次性加載到內(nèi)存：

# 分塊讀取 CSV（每次讀 1 萬(wàn)行）
chunk_iter = pd.read_csv('big_data.csv', chunksize=1000)

with sqlite3.connect('big_db.db') as conn:
    for chunk in chunk_iter:
        # 對(duì)每個(gè)塊做簡(jiǎn)單處理
        chunk['timestamp'] = pd.to_datetime(chunk['timestamp'])
        # 分塊寫(xiě)入數(shù)據(jù)庫(kù)
        chunk.to_sql(
            name='big_table',
            con=conn,
            if_exists='append',  # 追加模式
            index=False
        )
    print("全部寫(xiě)入完成！")

6 高級(jí)技巧：直接執(zhí)行 SQL 操作

Pandas 雖然強(qiáng)大，但復(fù)雜查詢(xún)?nèi)孕柚苯硬僮?SQL：

# 創(chuàng)建臨時(shí) DataFrame
df = pd.DataFrame({'product': ['A', 'B', 'C'], 'price': [10, 200, 150]})

# 寫(xiě)入 products 表
df.to_sql('products', conn, index=False, if_exists='replace')

# 執(zhí)行復(fù)雜查詢(xún)（連接 users 和 orders 表）
query = '''
    SELECT 
        u.name,
        p.product,
        p.price
    FROM users u
    JOIN orders o ON u.id = o.user_id
    JOIN products p ON o.product_id = p.id
    WHERE p.price > 10
'''
result_df = pd.read_sql(query, conn)
print(result_df)

7 避坑指南

數(shù)據(jù)類(lèi)型匹配問(wèn)題：

SQLite 默認(rèn)所有列為 TEXT，但 Pandas 會(huì)自動(dòng)推斷類(lèi)型。
寫(xiě)入時(shí)可用 dtype 參數(shù)手動(dòng)指定類(lèi)型：
```
df.to_sql('table', conn, dtype={'age': 'INTEGER', 'price': 'REAL'})
```

主鍵和索引：
- Pandas 不會(huì)自動(dòng)創(chuàng)建主鍵或索引，需提前用 SQL 語(yǔ)句定義表結(jié)構(gòu)。
性能瓶頸：
- 寫(xiě)入大量數(shù)據(jù)時(shí)，關(guān)閉事務(wù)自動(dòng)提交可提速：
```
with conn:
    df.to_sql(...)  # 使用上下文管理器自動(dòng)提交
```

8 總結(jié)

通過(guò) Pandas + SQLite3 的組合，你可以：
? 快速導(dǎo)入/導(dǎo)出數(shù)據(jù)：告別手動(dòng)拼接 SQL 語(yǔ)句。
? 無(wú)縫銜接數(shù)據(jù)分析：清洗、計(jì)算、可視化后直接入庫(kù)。
? 處理海量數(shù)據(jù)：分塊讀寫(xiě)避免內(nèi)存爆炸。

下一步建議：