Pandas & NumPy Academy · 课时

使用 chunksize 流式读取 CSV

使用 pd.read_csv(chunksize=) 按固定大小分块读取大型 CSV，处理每个数据块，再连接或累积结果。

使用 chunksize 流式读取 CSV 是 CoddyKit 上的免费 Pandas & NumPy Academy 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Pandas & NumPy Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Pandas & NumPy Academy 课程共包含 4 节课。

大型 CSV 文件的问题

当 CSV 文件大于可用 RAM 时——例如，一台内存为 16 GB 的计算机上有一个 50 GB 的日志文件——调用 pd.read_csv('file.csv') 会因 MemoryError 而失败，或者导致系统频繁交换，使运行速度慢到无法使用。解决方案是分块读取：不要一次性加载整个文件，而是将文件分成固定大小的部分进行处理，在不让所有数据同时驻留内存的情况下累积结果。

read_csv 中的 chunksize 参数

向 pd.read_csv() 传入 chunksize=N 后，返回的是 TextFileReader 迭代器，而不是 DataFrame。每次迭代最多生成 N 行的 DataFrame。文件会被延迟读取——在请求下一个数据块之前，不会加载任何数据。这个迭代器可以用于 for 循环，也可以传递给 pd.concat()。请选择足够大的 chunksize 以实现高效的输入输出（例如 10,000–100,000 行），同时又要足够小，以便轻松放入内存。

import pandas as pd

# Returns a TextFileReader iterator, NOT a DataFrame
chunks = pd.read_csv('sales_data.csv', chunksize=10000)
print(type(chunks))  # <class 'pandas.io.parsers.readers.TextFileReader'>

for chunk in chunks:
    print(f'Chunk shape: {chunk.shape}')
    # process each chunk independently
    break   # just show the first chunk here

独立处理每个数据块

最常见的模式是对每个数据块执行转换或筛选，将结果收集到列表中，然后进行拼接。例如，您可以筛选符合条件的行、计算每个数据块的统计信息，或只选择所需的列。在子集上工作意味着内存中只保留当前数据块，文件的其余部分不会被触碰。循环结束后，使用 pd.concat(results) 将结果组装成最终的 DataFrame。

import pandas as pd

results = []
for chunk in pd.read_csv('orders.csv', chunksize=50000):
    # Keep only high-value orders
    filtered = chunk[chunk['amount'] > 1000]
    results.append(filtered)

# Combine all filtered chunks
high_value = pd.concat(results, ignore_index=True)
print('High-value orders:', len(high_value))

跨数据块累积聚合结果

有时您根本不需要保留任何行，只需要一个持续聚合结果。您可以跨数据块跟踪累计和、计数或最小值/最大值，而无需不断构建 DataFrame 列表。这是内存效率最高的模式，因为无论文件大小如何，内存使用量都保持不变。最后，根据累加器计算最终统计量。

import pandas as pd

total_revenue = 0.0
total_rows = 0

for chunk in pd.read_csv('sales.csv', chunksize=100000):
    total_revenue += chunk['revenue'].sum()
    total_rows += len(chunk)

print(f'Processed {total_rows:,} rows')
print(f'Total revenue: ${total_revenue:,.2f}')

指定 dtype 以加快分块读取

默认情况下，Pandas 会根据数据推断列的数据类型，这需要扫描每个数据块两次（一次用于推断，一次用于解析）。提供 dtype 参数可以避免这项开销，还能防止不同数据块之间的数据类型不一致。例如，一个列中大多数值是整数，但有一个空单元格时，它在一个数据块中可能被推断为 float64，在另一个数据块中则被推断为 object。显式指定数据类型可以确保所有数据块读取时类型一致且速度更快。

import pandas as pd

dtype_map = {
    'order_id': 'int32',
    'customer_id': 'int32',
    'amount': 'float32',
    'category': 'category'
}

for chunk in pd.read_csv('orders.csv',
                         chunksize=50000,
                         dtype=dtype_map,
                         parse_dates=['order_date']):
    print(chunk.dtypes)
    break

仅选择所需的列

使用 usecols 参数只加载分析所需的列。如果一个 CSV 有 50 列，而您的聚合只使用其中 3 列，就没有理由解析另外 47 列。将 usecols 与 chunksize 结合使用，可以大幅减少输入输出时间和内存使用量。这是处理大型 CSV 时最简单、效果也最显著的优化方法之一。

import pandas as pd

# Only read the three columns we actually need
for chunk in pd.read_csv(
    'large_transactions.csv',
    chunksize=100000,
    usecols=['date', 'amount', 'region']
):
    print(chunk.columns.tolist())
    print(chunk.memory_usage(deep=True).sum() / 1e6, 'MB per chunk')
    break

数据块中的 GroupBy 聚合

跨数据块执行 groupby 聚合需要累积部分结果。先在每个数据块中计算 groupby，然后将部分结果拼接起来，再使用第二个 groupby 进行合并。例如，要计算一个 10 GB 文件中各地区的总销售额，可以将每个数据块的地区销售额总和收集到列表中，然后拼接并再次分组。这种两遍聚合模式有时也称为映射-归约方法。

import pandas as pd

partials = []
for chunk in pd.read_csv('sales.csv',
                         chunksize=100000,
                         usecols=['region', 'revenue']):
    partial = chunk.groupby('region')['revenue'].sum()
    partials.append(partial)

# Combine partial sums
final = pd.concat(partials).groupby(level=0).sum()
print('Revenue by region:')
print(final.sort_values(ascending=False))

处理跨数据块的解析错误

来自外部来源的大型 CSV 文件通常包含格式错误的行——例如多余的逗号、错误的编码或被截断的行。可以使用 on_bad_lines='skip'（Pandas 1.3 及更高版本）或 error_bad_lines=False（较旧版本的 Pandas）静默跳过错误行；如果 UTF-8 解析失败，则使用 encoding='latin-1'。在处理每个数据块时使用 try-except，跟踪哪些数据块产生了错误，从而构建一个健壮的管道，不会因一个包含 1,000 万行的文件中的单行错误而崩溃。

import pandas as pd

bad_chunks = []
all_chunks = []

for i, chunk in enumerate(pd.read_csv(
    'raw_data.csv',
    chunksize=50000,
    on_bad_lines='skip',
    encoding='utf-8',
    encoding_errors='replace'
)):
    try:
        # Your transformation here
        all_chunks.append(chunk)
    except Exception as e:
        bad_chunks.append((i, str(e)))
        print(f'Chunk {i} error: {e}')

print(f'Processed {len(all_chunks)} chunks, {len(bad_chunks)} errors')

分块写入输出文件

当处理后的输出同样很大时，应增量写入结果，而不是将所有内容累积到内存中后最后一次性写入。打开一个 CSV 文件，并使用 mode='a' 追加每个已处理的数据块；对于后续数据块，使用 header=False。这样可以使输出管道的内存使用量保持恒定，并允许您在完整运行结束前检查部分结果。

import pandas as pd

first_chunk = True
for chunk in pd.read_csv('input.csv', chunksize=100000):
    # Transform
    processed = chunk[chunk['status'] == 'active'].copy()
    processed['revenue_usd'] = processed['revenue'] * 1.10

    # Write incrementally
    mode = 'w' if first_chunk else 'a'
    processed.to_csv('output.csv',
                     mode=mode,
                     header=first_chunk,
                     index=False)
    first_chunk = False

print('Done writing output.csv')

估算最佳数据块大小

选择 chunksize 需要权衡：太小会导致 Python 循环迭代次数过多和开销较高；太大则可能无法放入 RAM。一种实用方法是加载一个数据块，使用 chunk.memory_usage(deep=True).sum() 测量其内存占用，并设置 chunksize，使每个数据块大约使用可用 RAM 的 10–20%。Python 的 psutil.virtual_memory().available 可以在运行时提供可用 RAM，从而支持自适应计算 chunksize。

import pandas as pd

# Sample 1000 rows to estimate per-row memory
sample = pd.read_csv('big_file.csv', nrows=1000)
bytes_per_row = sample.memory_usage(deep=True).sum() / 1000
print(f'Bytes per row: {bytes_per_row:.0f}')

# Target: use at most 500 MB per chunk
target_bytes = 500 * 1024 * 1024
optimal_chunksize = int(target_bytes / bytes_per_row)
print(f'Recommended chunksize: {optimal_chunksize:,}')

高效合并分块结果

将许多数据块 DataFrame 累积到列表中再进行拼接时，请注意：对数百个小型数据框调用 pd.concat 会因反复分配内存而运行缓慢。更好的模式是在每个数据块内进行聚合，只存储较小的聚合结果，而不是完整的数据块。如果确实需要保留所有行，使用 pyarrow 增量写入 Parquet 文件，比最后调用 pd.concat 更快。

import pandas as pd

# Efficient: aggregate first, small list of scalars
running_total = 0
running_count = 0

for chunk in pd.read_csv('sales.csv', chunksize=100000):
    running_total += chunk['amount'].sum()
    running_count += chunk['amount'].count()

print(f'Mean amount: {running_total / running_count:.2f}')

# Avoid: accumulating full chunk DataFrames
# results = []
# for chunk in reader:
#     results.append(chunk)   # memory grows to full file size
# df = pd.concat(results)     # slow for hundreds of chunks

快速检查

测试您对本课数据分析概念的理解。

课程回顾

在本课中，您学习了：pd.read_csv 中的分块大小会返回一个 DataFrame 迭代器，从而支持以内存高效的方式处理大型文件；usecols 和 dtype参数可以减少每个数据块的内存占用并加快解析；以及持续累加器（总和、计数、部分结果）可以避免构建包含所有数据块的列表。接下来，我们将更深入地学习跨数据块的增量聚合模式。

常见问题解答

「使用 chunksize 流式读取 CSV」课时是免费的吗？

是的 — 「使用 chunksize 流式读取 CSV」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Pandas & NumPy Academy 课程的其余内容，请升级到 CoddyKit PRO。 Pandas & NumPy Academy 课程共包含 4 节课。

「使用 chunksize 流式读取 CSV」这节课中我会学到什么？

使用 pd.read_csv(chunksize=) 按固定大小分块读取大型 CSV，处理每个数据块，再连接或累积结果。你通过在浏览器中直接运行的动手代码来练习 Pandas & NumPy Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Pandas & NumPy Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Pandas & NumPy Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「使用 chunksize 流式读取 CSV」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Pandas & NumPy Academy 课中编写并运行代码吗？

能。每节 Pandas & NumPy Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 Pandas & NumPy Academy