4種Pandas高效讀取文件數(shù)據(jù)的完整方法指南

更新時間：2025年12月16日 09:50:58 作者：用戶6854537597769

這篇文章主要為大家詳細(xì)介紹了4種Pandas高效讀取文件數(shù)據(jù)的完整方法,包括CSV文件,Excel文件,JSON文件和SQL,文中的示例代碼講解詳細(xì),有需要的小伙伴可以了解下

痛點暴擊開頭

你有沒有遇到過這種情況：

剛拿到一個1GB的CSV文件，你雙擊打開...Excel直接卡死，電腦風(fēng)扇狂轉(zhuǎn)，你只能強制重啟電腦。

或者更慘：你寫了pd.read_csv('huge_file.csv')，然后去接了杯水，刷了會兒朋友圈，回來一看...內(nèi)存爆了，Python進(jìn)程被強制殺死。

上周我們組來了個實習(xí)生，處理一個銷售數(shù)據(jù)的Excel文件，他老老實實地一個sheet一個sheet地復(fù)制粘貼，結(jié)果花了一整天還沒弄完。

老大看了直搖頭："你知道pandas可以一鍵讀取所有sheet嗎？"

實習(xí)生愣住了。我也愣住了。

今天咱們就來徹底搞懂pandas的數(shù)據(jù)讀取，讓你從小白秒變數(shù)據(jù)讀取高手！

CSV讀取：不是所有逗號都是平等的

錯誤示范（血的教訓(xùn)）

# ? 新手這樣寫（包括三個月前的我）
import pandas as pd

# 簡單粗暴地讀取
df = pd.read_csv('data.csv')

# 遇到中文亂碼？涼拌！
# 遇到分號分隔？涼拌！
# 遇到?jīng)]有表頭？涼拌！

結(jié)果呢？要么中文變成亂碼，要么列名錯位，要么直接報錯。就像你女朋友說"隨便"，然后你真的隨便了...

正確姿勢：優(yōu)雅地處理各種奇葩CSV

# ? 專業(yè)人士的打開方式
import pandas as pd

# 基礎(chǔ)但完整的讀取
df = pd.read_csv('data.csv',
                encoding='utf-8',        # 解決中文亂碼
                sep=',',                 # 明確分隔符
                header=0,                # 表頭在第幾行
                index_col=0,             # 哪一列作為索引
                na_values=['NA', 'null', 'N/A'])  # 空值怎么識別

但是，真實世界的CSV文件遠(yuǎn)比你想象的要奇葩！

# ?? 高手進(jìn)階：處理各種奇葩情況

# 1. 分號分隔的歐洲格式
df_euro = pd.read_csv('euro_data.csv', sep=';')

# 2. 沒有表頭，需要自己命名
df_no_header = pd.read_csv('data_no_header.csv',
                          header=None,
                          names=['ID', 'Name', 'Age', 'Score'])

# 3. 表頭不在第一行（比如某些導(dǎo)出文件）
df_skip = pd.read_csv('report.csv', header=5)  # 跳過前5行，第6行作為表頭

# 4. 自動識別分隔符（遇到奇葩文件時）
df_auto = pd.read_csv('weird_file.csv', sep=None, engine='python')

# 5. 只讀取需要的列（內(nèi)存優(yōu)化利器）
df_cols = pd.read_csv('big_data.csv',
                     usecols=['name', 'age', 'salary'])  # 只要這3列

# 6. 分塊讀取超大文件（內(nèi)存不夠時的救星）
chunk_iter = pd.read_csv('huge_file.csv', chunksize=10000)
for chunk in chunk_iter:
    # 每次處理10000行
    process_chunk(chunk)

關(guān)鍵參數(shù)解析：

encoding：utf-8、gbk、latin-1，解決亂碼的神器
sep：分隔符，可以是逗號、分號、制表符等
header：表頭行號，None表示沒有表頭
usecols：只讀取指定列，節(jié)省內(nèi)存的利器
chunksize：分塊讀取，處理大文件的法寶

Excel讀取：告別手動復(fù)制粘貼

錯誤示范：手動操作的地獄

# ? 低效到哭的操作
# 1. 手動打開Excel
# 2. 一個sheet一個sheet地復(fù)制
# 3. 粘貼到Python里
# 4. 重復(fù)以上步驟100次...

# 更慘的情況：
# - Excel文件太大，打開就要5分鐘
# - 有密碼保護(hù)，還要輸密碼
# - 包含公式，復(fù)制過來全是錯誤值

正確姿勢：一行代碼讀取所有sheet

# ? 優(yōu)雅的Excel讀取
import pandas as pd

# 讀取單個sheet
df = pd.read_excel('sales_data.xlsx', sheet_name='Sheet1')

# ?? 高手操作：一次性讀取所有sheet
all_sheets = pd.read_excel('multi_sheet.xlsx', sheet_name=None)
# 結(jié)果是一個字典，key是sheet名，value是對應(yīng)的DataFrame

for sheet_name, df in all_sheets.items():
    print(f"Sheet: {sheet_name}, 行數(shù): {len(df)}")

# ?? 指定讀取第幾個sheet（從0開始）
df_second = pd.read_excel('data.xlsx', sheet_name=1)

# ?? 指定引擎（解決某些兼容性問題）
df_xls = pd.read_excel('old_file.xls', engine='openpyxl')
df_xlsx = pd.read_excel('new_file.xlsx', engine='xlrd')

處理Excel的特殊情況

# ?? 處理復(fù)雜的Excel文件

# 1. 指定讀取行范圍
df_partial = pd.read_excel('data.xlsx',
                          sheet_name='Sales',
                          skiprows=5,       # 跳過前5行
                          nrows=100,        # 只讀取100行
                          usecols='A:E')    # 只讀取A到E列

# 2. 處理日期列
df_dates = pd.read_excel('data.xlsx',
                        parse_dates=['訂單日期', '發(fā)貨日期'])

# 3. 處理多個表頭（復(fù)雜的報表格式）
df_multi_header = pd.read_excel('complex_report.xlsx',
                               header=[0, 1])  # 前兩行都是表頭

# 4. 讀取指定單元格區(qū)域
# 先讀取整個sheet，再切片
df_full = pd.read_excel('data.xlsx', sheet_name='Data')
df_subset = df_full.iloc[10:50, 2:8]  # 第11-50行，第3-8列

JSON讀取：告別eval()的危險操作

錯誤示范：玩火的行為

# ? 危險操作！千萬別這么干！
import json
import pandas as pd

# 有人居然用eval()...
with open('data.json', 'r') as f:
    data_str = f.read()
    data = eval(data_str)  # ?? 危險！可以執(zhí)行任意代碼！

# 或者更騷的操作：手動解析
# 結(jié)果：錯漏百出，維護(hù)困難

正確姿勢：安全優(yōu)雅的JSON處理

# ? 專業(yè)人士的JSON讀取
import pandas as pd
from io import StringIO

# 1. 讀取JSON文件
df = pd.read_json('data.json')

# 2. 從字符串讀取JSON
json_str = '{"name": ["張三", "李四"], "age": [25, 30]}'
df = pd.read_json(StringIO(json_str))

# ?? 處理不同的JSON格式

# 3. 嵌套JSON的扁平化
nested_json = '''
{
    "employees": [
        {"name": "張三", "dept": {"name": "技術(shù)部", "id": 1}},
        {"name": "李四", "dept": {"name": "銷售部", "id": 2}}
    ]
}
'''
df_nested = pd.read_json(StringIO(nested_json))

# 4. 行格式JSON（每行一個JSON對象）
# 適用于日志文件、API返回等場景
lines_json = '''
{"name": "張三", "age": 25, "city": "北京"}
{"name": "李四", "age": 30, "city": "上海"}
{"name": "王五", "age": 28, "city": "廣州"}
'''
df_lines = pd.read_json(StringIO(lines_json), lines=True)

# 5. 指定JSON的讀取方向
df_records = pd.read_json('data.json', orient='records')  # 記錄格式
df_index = pd.read_json('data.json', orient='index')     # 索引格式
df_values = pd.read_json('data.json', orient='values')   # 值格式

JSON數(shù)據(jù)清洗實戰(zhàn)

# ?? 真實場景：處理API返回的JSON數(shù)據(jù)
import requests
import pandas as pd

# 模擬API調(diào)用
response = requests.get('https://api.example.com/users')
data = response.json()

# 轉(zhuǎn)換為DataFrame
df_users = pd.DataFrame(data['users'])

# 嵌套字段展開
df_users['city'] = df_users['address'].apply(lambda x: x['city'])
df_users['street'] = df_users['address'].apply(lambda x: x['street'])

# 刪除原始嵌套字段
df_users = df_users.drop('address', axis=1)

# 日期格式轉(zhuǎn)換
df_users['created_at'] = pd.to_datetime(df_users['created_at'])

SQL讀?。簲?shù)據(jù)庫連接的藝術(shù)

錯誤示范：手動導(dǎo)出的痛苦

# ? 傳統(tǒng)但低效的方式
# 1. 打開數(shù)據(jù)庫管理工具
# 2. 寫SQL查詢
# 3. 導(dǎo)出為CSV
# 4. 用pandas讀取CSV
# 5. 發(fā)現(xiàn)數(shù)據(jù)不對，重復(fù)以上步驟...

# 或者更慘：
# - 手動復(fù)制粘貼結(jié)果，超過1000行就卡死
# - 導(dǎo)出的文件編碼不對，中文亂碼
# - 每次都要重復(fù)導(dǎo)出，煩死人

正確姿勢：直接連接數(shù)據(jù)庫

# ? 優(yōu)雅的數(shù)據(jù)庫讀取
import pandas as pd
import sqlite3  # SQLite示例
from sqlalchemy import create_engine  # 更通用的方案

# 1. SQLite連接（輕量級，文件數(shù)據(jù)庫）
conn = sqlite3.connect('database.db')

# 簡單查詢
df = pd.read_sql_query("SELECT * FROM users WHERE age > 25", conn)

# ?? 高級操作

# 2. 使用SQLAlchemy（支持各種數(shù)據(jù)庫）
# MySQL: engine = create_engine('mysql://user:password@host:port/database')
# PostgreSQL: engine = create_engine('postgresql://user:password@host:port/database')
engine = create_engine('sqlite:///database.db')

# 查詢并直接返回DataFrame
df = pd.read_sql("SELECT * FROM sales WHERE date >= '2023-01-01'", engine)

# 3. 讀取整個表
df_table = pd.read_sql_table('users', engine)

# 4. 使用參數(shù)化查詢（防SQL注入）
query = "SELECT * FROM products WHERE price > ? AND category = ?"
df_params = pd.read_sql(query, engine, params=[100, 'Electronics'])

# 5. 分塊讀取大表
chunk_iter = pd.read_sql("SELECT * FROM large_table", engine, chunksize=10000)
for chunk in chunk_iter:
    process_chunk(chunk)

SQL讀取的最佳實踐

# ?? 生產(chǎn)環(huán)境的SQL讀取方案

# 1. 連接池配置（提高性能）
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    'postgresql://user:password@localhost:5432/mydb',
    poolclass=QueuePool,
    pool_size=5,
    max_overflow=10
)

# 2. 上下文管理器（確保連接關(guān)閉）
with engine.connect() as conn:
    df = pd.read_sql("SELECT * FROM orders", conn)

# 3. 復(fù)雜查詢的優(yōu)化
# 只讀取需要的列
df_optimized = pd.read_sql("""
    SELECT id, name, email, created_at
    FROM users
    WHERE status = 'active'
    ORDER BY created_at DESC
    LIMIT 1000
""", engine)

# 4. 處理大表的數(shù)據(jù)類型
# 避免內(nèi)存溢出，指定數(shù)據(jù)類型
dtypes = {
    'user_id': 'int32',
    'price': 'float32',
    'status': 'category'
}
df_efficient = pd.read_sql("SELECT * FROM transactions", engine, dtype=dtypes)

# 5. 時間處理
df_time = pd.read_sql("""
    SELECT *,
           EXTRACT(YEAR FROM created_at) as year,
           EXTRACT(MONTH FROM created_at) as month
    FROM orders
""", engine, parse_dates=['created_at'])

實戰(zhàn)演練：一個完整的數(shù)據(jù)處理流程

假設(shè)你收到了一份銷售數(shù)據(jù)，包含CSV、Excel、JSON和數(shù)據(jù)庫四種格式：

import pandas as pd
import sqlite3
from pathlib import Path

# ?? 完整的數(shù)據(jù)處理示例
def process_sales_data():

    # 1. 讀取主數(shù)據(jù)（CSV）
    df_sales = pd.read_csv('sales_main.csv',
                          encoding='utf-8',
                          parse_dates=['date'],
                          dtype={'product_id': 'str', 'customer_id': 'str'})

    # 2. 讀取產(chǎn)品信息（Excel）
    excel_data = pd.read_excel('products.xlsx', sheet_name=None)
    df_products = excel_data['ProductList']
    df_categories = excel_data['Categories']

    # 3. 讀取客戶信息（JSON）
    df_customers = pd.read_json('customers.json')

    # 4. 讀取補充數(shù)據(jù)（數(shù)據(jù)庫）
    conn = sqlite3.connect('supplementary.db')
    df_regions = pd.read_sql_query("SELECT * FROM regions", conn)

    # 5. 數(shù)據(jù)合并和分析
    # 合并銷售數(shù)據(jù)和產(chǎn)品信息
    df_merged = df_sales.merge(df_products, on='product_id', how='left')

    # 合并客戶信息
    df_merged = df_merged.merge(df_customers, on='customer_id', how='left')

    # 合并地區(qū)信息
    df_merged = df_merged.merge(df_regions, on='region_code', how='left')

    # 6. 數(shù)據(jù)清洗和轉(zhuǎn)換
    # 處理缺失值
    df_merged['price'] = df_merged['price'].fillna(df_merged['price'].median())

    # 計算銷售金額
    df_merged['total_amount'] = df_merged['quantity'] * df_merged['price']

    # 提取月份
    df_merged['month'] = df_merged['date'].dt.month

    return df_merged

# 使用函數(shù)
df_result = process_sales_data()
print(f"處理完成！共{len(df_result)}條記錄")

性能優(yōu)化：讓你的數(shù)據(jù)讀取飛起來

內(nèi)存優(yōu)化技巧

# ?? 大文件處理優(yōu)化方案

# 1. 指定數(shù)據(jù)類型（減少內(nèi)存占用）
dtypes_optimized = {
    'id': 'int32',
    'name': 'category',  # 重復(fù)值多的列用category類型
    'price': 'float32',
    'is_active': 'bool'  # 布爾值
}

df_small = pd.read_csv('large_file.csv', dtype=dtypes_optimized)

# 2. 只讀取需要的列
df_selected = pd.read_csv('data.csv',
                         usecols=['name', 'age', 'salary'],
                         dtype={'name': 'str', 'age': 'int16', 'salary': 'float32'})

# 3. 分塊處理大文件
chunk_size = 100000
results = []

for chunk in pd.read_csv('huge_file.csv', chunksize=chunk_size):
    # 對每個chunk進(jìn)行處理
    processed_chunk = chunk[chunk['salary'] > 50000]
    results.append(processed_chunk)

df_final = pd.concat(results, ignore_index=True)

# 4. 使用低內(nèi)存引擎
df_low_memory = pd.read_csv('data.csv', engine='c', low_memory=True)

并行讀取優(yōu)化

# ?? 并行處理多個文件
import pandas as pd
from concurrent.futures import ThreadPoolExecutor
import glob

def read_file(file_path):
    """讀取單個文件的函數(shù)"""
    try:
        return pd.read_csv(file_path)
    except Exception as e:
        print(f"讀取文件 {file_path} 失敗: {e}")
        return None

# 獲取所有CSV文件
file_list = glob.glob('data/*.csv')

# 使用線程池并行讀取
with ThreadPoolExecutor(max_workers=4) as executor:
    dfs = list(executor.map(read_file, file_list))

# 過濾掉讀取失敗的文件
dfs = [df for df in dfs if df is not None]

# 合并所有DataFrame
df_combined = pd.concat(dfs, ignore_index=True)
print(f"成功讀取并合并了{(lán)len(dfs)}個文件")

常見坑爹問題及解決方案

1. 編碼問題

# ? 編碼錯誤：UnicodeDecodeError
# df = pd.read_csv('chinese_data.csv')  # 可能報錯

# ? 解決方案
try:
    df = pd.read_csv('chinese_data.csv', encoding='utf-8')
except UnicodeDecodeError:
    try:
        df = pd.read_csv('chinese_data.csv', encoding='gbk')
    except UnicodeDecodeError:
        df = pd.read_csv('chinese_data.csv', encoding='latin-1')

2. 內(nèi)存不足

# ? 內(nèi)存溢出：MemoryError
# df = pd.read_csv('10gb_file.csv')

# ? 解決方案：分塊讀取
chunk_size = 50000
chunk_iter = pd.read_csv('10gb_file.csv', chunksize=chunk_size)

for i, chunk in enumerate(chunk_iter):
    # 處理每個chunk，比如保存到多個小文件
    chunk.to_csv(f'processed_chunk_{i}.csv', index=False)
    print(f"處理完成第{i+1}個chunk")

3. 日期時間解析

# ? 日期被當(dāng)作字符串讀取
# df = pd.read_csv('data.csv')

# ? 正確處理日期
df = pd.read_csv('data.csv',
                parse_dates=['date_col'],  # 指定日期列
                date_parser=lambda x: pd.datetime.strptime(x, '%Y-%m-%d'))

# 或者讀取后再轉(zhuǎn)換
df['date_col'] = pd.to_datetime(df['date_col'])