从基础到高级详解Python高效读写CSV数据实战技巧

import csv

with open('data.csv', 'r', newline='', encoding='utf-8') as file:
    reader = csv.reader(file)
    for row in reader:
        print(row)

with open('data.csv', 'r', newline='', encoding='utf-8') as file:
    reader = csv.DictReader(file)
    for row in reader:
        print(row['name'], row['age'], row['city'])

import csv

data = [
    ["姓名", "年龄", "城市"],
    ["Alice", 25, "New York"],
    ["Bob", 30, "Los Angeles"],
    ["Charlie", 35, "Chicago"]
]

with open('output.csv', 'w', newline='', encoding='utf-8') as file:
    writer = csv.writer(file)
    writer.writerows(data)

fieldnames = ['Name', 'Age', 'City']
data = [
    {'Name': 'Alice', 'Age': 24, 'City': 'New York'},
    {'Name': 'Bob', 'Age': 30, 'City': 'Los Angeles'}
]

极速分析

with open('output.csv', 'w', newline='', encoding='utf-8') as file:
    writer = csv.DictWriter(file, fieldnames=fieldnames)
    writer.writeheader()  # 写入标题行
    writer.writerows(data)

import pandas as pd

df = pd.read_csv('data.csv')
print(df.head())  # 查看前几行数据

# 处理缺失值
df.fillna(0, inplace=True)  # 将所有缺失值替换为0
# 或者用特定列的平均值替换
df['column1'].fillna(df['column1'].mean(), inplace=True)

# 删除含有缺失值的行
df.dropna(inplace=True)

# 数据类型转换
df['age'] = df['age'].astype(int)

# 数据过滤
filtered_df = df[df['age'] > 30]  # 筛选年龄大于30的行

# 分组操作
grouped = df.groupby('city')['age'].mean()  # 按城市分组极速分析计算平均年龄

df.to_csv('output.csv', index=False)  # index=False表示不写入行索引

# 分批写入
for i in range(0, len(df), 1000):
    chunk = df[i:i+1000]
    chunk.to_csv(f'output_chunk_{i}.csv', index=False)
    
# 使用压缩
df.to_csv('output.csv.gz', index=False, compression='gzip')

# 使用csv模块逐行处理（内存友好）
with open('large_data.csv', 'r') as file:
    reader = csv.reader(file)
    for极速分析 row in reader:
        process_row(row)  # 自定义处理函数

# 使用pandas分块读取
chunk_size = 10000
for chunk in pd.read_csv('large_data.csv', chunksize=chunk_size):
    process_chunk(chunk)  # 处理每个数据块

# 读取制表符分隔的文件
with open('data.ts极速分析v', 'r') as file:
    reader = csv.reader(file, delimiter='\t')
    for row in reader:
        print(row)

# 处理包含特殊字符的字段
with open('data.csv', 'w', newline='') as file:
    writer = csv.writer(file, quoting=csv.QUOTE_NONNUMERIC)  # 非数字字段加引号
    writer.writerow(['Text', 100, 'More "text"'])

# 手动指定数据类型以减少内存使用和提高处理速度
dtypes = {
    'name': 'category',
    'age': 'int32',
    'salary': 'float32'
}
df = pd.read_csv('data.csv', dtype=dtypes)

# 在读取时解析日期
df = pd.read_csv('data.csv', parse_dates=['birth_date'])

# 使用ciso8601快速解析日期时间
from ciso8601 import parse_datetime
date_string = "2023-10-01T12:00:00"
parsed_date = parse_datetime(date_string)

# 使用cython编译关键代码段
# 需要创建setup.py文件和编写cython代码

# 读取UTF-8编码的文件
with open('data.csv', 'r', encoding='utf-8') as file:
    reader = csv.reader(file)
    for row in reader:
        print(row)

# 读取GBK编码的文件（常见于中文环境）
with open('data.csv', 'r', encoding='gbk') as file:
    reader = csv.reader(file)
    for row in reader:
        print(row)

# 跳过错误行（使用pandas）
try:
    df = pd.read_csv('data.csv', error_bad_lines=False)
except Exception as e:
    print(f"Error: {e}")

# 使用灵活的分隔符和引号处理
import csv
with open('messy_data.csv', 'r') as file:
    # 尝试多种分隔符
    dialects = csv.Sniffer().sniff(file.read(1024))
    file.seek(0)
    reader = csv.reader(file, delimiter=dialects.delimiter)
    for row in reader:
        print(row)