跨数据块增量聚合
跨多个数据块累积运行中的计数、总和以及最小值/最大值，而无需将完整文件存入内存。
跨数据块增量聚合 是 CoddyKit 上的免费 Pandas & NumPy Academy 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 Pandas & NumPy Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Pandas & NumPy Academy 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
Why Incremental Aggregation?
Incremental aggregation is the key to analysing datasets larger than RAM without distributing computation across multiple machines. Instead of loading all data to compute a final statistic, you maintain running accumulators — partial sums, counts, min/max values — updating them with each chunk. The final result is assembled from these lightweight accumulators after the full file is scanned. This pattern scales to terabyte files on a single laptop.
Counting Rows and Computing Mean
Computing the mean across chunks requires tracking the running sum and count separately. You cannot simply average the per-chunk means because chunks may have different sizes. The correct formula is total_sum / total_count. This pattern extends to any quantity that can be decomposed: variance, correlation, and histograms all have incremental formulas.
import pandas as pd
total_sum = 0.0
total_count = 0
for chunk in pd.read_csv('transactions.csv', chunksize=100000):
total_sum += chunk['amount'].sum()
total_count += chunk['amount'].notna().sum()
grand_mean = total_sum / total_count
print(f'Rows processed: {total_count:,}')
print(f'Grand mean: {grand_mean:.4f}')Incremental Min and Max
Tracking the global minimum and maximum across chunks is straightforward: initialise with Python's float('inf') and float('-inf'), then update with each chunk's min/max. This avoids any intermediate storage. The pattern generalises to per-group min/max by maintaining a dictionary keyed by group identifier.
import pandas as pd
global_min = float('inf')
global_max = float('-inf')
for chunk in pd.read_csv('prices.csv', chunksize=50000):
chunk_min = chunk['price'].min()
chunk_max = chunk['price'].max()
if chunk_min < global_min:
global_min = chunk_min
if chunk_max > global_max:
global_max = chunk_max
print(f'Price range: {global_min} to {global_max}')Incremental Frequency Counts
For categorical columns, maintain a running frequency dictionary by adding each chunk's value_counts() result to a Pandas Series accumulator. Because Pandas Series addition aligns on index labels, unknown categories in later chunks are automatically included. After all chunks, sort by count to see the top categories across the entire dataset.
import pandas as pd
freq = pd.Series(dtype='int64')
for chunk in pd.read_csv('orders.csv',
chunksize=100000,
usecols=['category']):
chunk_counts = chunk['category'].value_counts()
freq = freq.add(chunk_counts, fill_value=0)
# Final sorted frequency table
print(freq.sort_values(ascending=False).head(10))Incremental GroupBy Aggregation
To compute a groupby sum or count across chunks, apply groupby().agg() within each chunk and store the resulting Series or DataFrame. After the loop, concatenate all partial results and apply a second groupby to combine them. This two-stage approach correctly handles groups that appear in multiple chunks, which is common when data is sorted by date rather than by group.
import pandas as pd
partials = []
for chunk in pd.read_csv('sales.csv',
chunksize=100000,
usecols=['region', 'product', 'revenue']):
p = chunk.groupby(['region', 'product'])['revenue'].sum()
partials.append(p)
final = (
pd.concat(partials)
.groupby(level=['region', 'product'])
.sum()
.sort_values(ascending=False)
)
print(final.head(10))Computing Variance Incrementally (Welford's Method)
Computing variance across chunks is trickier than mean. The naive formula E[X²] - E[X]² suffers from catastrophic cancellation for large means. Welford's online algorithm maintains a running mean and sum-of-squared-deviations, updating them with each new value in a numerically stable way. While SciPy implements this, understanding the pattern lets you extend it to weighted variance and covariance.
import pandas as pd
import numpy as np
# Simple two-pass approach using stored chunk stats
chunk_stats = []
for chunk in pd.read_csv('data.csv',
chunksize=100000,
usecols=['value']):
n = chunk['value'].count()
mean = chunk['value'].mean()
var = chunk['value'].var(ddof=1)
chunk_stats.append((n, mean, var))
# Combine: use pooled variance formula
total_n = sum(s[0] for s in chunk_stats)
total_mean = sum(s[0]*s[1] for s in chunk_stats) / total_n
pooled_var = sum((s[0]-1)*s[2] + s[0]*(s[1]-total_mean)**2
for s in chunk_stats) / (total_n - 1)
print(f'Grand variance: {pooled_var:.4f}')Building an Incremental Histogram
Computing the distribution of a column across a file too large for RAM requires an incremental histogram. Fix the bin edges upfront (based on a small sample or domain knowledge), then use np.histogram(chunk_values, bins=edges) within each chunk and accumulate the counts. At the end, plot the combined counts as a bar chart. This is how streaming systems like Kafka Streams and Flink compute approximate histograms.
import pandas as pd
import numpy as np
# Decide bin edges from a sample
sample = pd.read_csv('amounts.csv', nrows=5000)
bins = np.linspace(sample['amount'].min(),
sample['amount'].max(), 21)
counts = np.zeros(len(bins) - 1, dtype='int64')
for chunk in pd.read_csv('amounts.csv',
chunksize=100000,
usecols=['amount']):
chunk_counts, _ = np.histogram(
chunk['amount'].dropna(), bins=bins
)
counts += chunk_counts
print('Histogram counts:', counts[:5], '...')Tracking Unique Values Approximately
Counting exact distinct values across chunks requires storing all unique values — potentially millions. For approximate counts at scale, use a HyperLogLog sketch, available in Python via the hyperloglog library. Alternatively, track uniques per chunk with a set and take the union, but this grows unboundedly. For a cheap approximation, use pd.Series.nunique() per chunk and report the average — not exact, but often sufficient for data profiling.
import pandas as pd
unique_ids = set()
for chunk in pd.read_csv('events.csv',
chunksize=100000,
usecols=['user_id']):
unique_ids.update(chunk['user_id'].dropna().unique())
print(f'Distinct user IDs: {len(unique_ids):,}')
# Warning: the set may grow large for high-cardinality columnsProgress Reporting During Long Runs
Processing a multi-gigabyte file can take minutes. Add progress reporting so you know the pipeline is running and can estimate remaining time. Count bytes or rows processed and compare against the file size. The tqdm library makes this trivial with its tqdm(reader) wrapper. Even without tqdm, printing a status line every 10 chunks gives valuable feedback during long batch jobs.
import pandas as pd
import time
chunksize = 100000
start = time.time()
rows_processed = 0
for i, chunk in enumerate(pd.read_csv('big.csv',
chunksize=chunksize)):
rows_processed += len(chunk)
# Report every 10 chunks
if (i + 1) % 10 == 0:
elapsed = time.time() - start
rate = rows_processed / elapsed
print(f'Chunk {i+1}: {rows_processed:,} rows '
f'@ {rate/1000:.0f}k rows/sec')
print(f'Total: {rows_processed:,} rows in {time.time()-start:.1f}s')Filtering Before Aggregating
Apply filters within each chunk before aggregating to avoid accumulating unwanted data. For example, if you only care about 2024 orders, filter the chunk's date column before groupby. This reduces the memory required for partial results and speeds up the final concatenation step. Always push filters as early as possible in the pipeline — a fundamental principle of efficient data processing.
import pandas as pd
partials = []
for chunk in pd.read_csv('orders.csv',
chunksize=100000,
parse_dates=['order_date']):
# Filter early: only 2024 orders
mask = chunk['order_date'].dt.year == 2024
filtered = chunk.loc[mask, ['category', 'revenue']]
if len(filtered) > 0:
p = filtered.groupby('category')['revenue'].sum()
partials.append(p)
if partials:
result = pd.concat(partials).groupby(level=0).sum()
print(result)Saving Intermediate Results
For very long-running jobs, save intermediate results periodically so you can resume from a checkpoint if the process is interrupted. Write per-chunk aggregates to a Parquet or CSV file after every N chunks. If the job fails at chunk 800 of 1000, you can reload the saved aggregates and continue from where you left off rather than reprocessing the entire file. This resilience pattern is essential in production data pipelines.
import pandas as pd
import os
CHECKPOINT = 'checkpoint.csv'
running_total = 0.0
running_count = 0
# Resume from checkpoint if it exists
if os.path.exists(CHECKPOINT):
ckpt = pd.read_csv(CHECKPOINT)
running_total = ckpt['total'].iloc[0]
running_count = int(ckpt['count'].iloc[0])
print(f'Resuming from checkpoint: {running_count:,} rows')
for chunk in pd.read_csv('huge.csv', chunksize=100000):
running_total += chunk['value'].sum()
running_count += len(chunk)
# Save checkpoint
pd.DataFrame({'total': [running_total],
'count': [running_count]}).to_csv(CHECKPOINT, index=False)
print(f'Final mean: {running_total / running_count:.4f}')Quick Check
Test your understanding of Data Analysis concepts from this lesson.
Lesson Recap
In this lesson you learned: running accumulators (sum, count, min/max, frequency Series) enable constant-memory aggregation across large files, two-stage groupby (partial groupby per chunk, then concat and regroup) handles cross-chunk groups correctly, and early filtering within each chunk reduces the cost of the accumulation step. Next up we explore Dask DataFrames as a drop-in parallel replacement for Pandas on large datasets.
常见问题解答
「跨数据块增量聚合」课时是免费的吗？
是的 — 「跨数据块增量聚合」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Pandas & NumPy Academy 课程的其余内容，请升级到 CoddyKit PRO。 Pandas & NumPy Academy 课程共包含 4 节课。
「跨数据块增量聚合」这节课中我会学到什么？
跨多个数据块累积运行中的计数、总和以及最小值/最大值，而无需将完整文件存入内存。 你通过在浏览器中直接运行的动手代码来练习 Pandas & NumPy Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 Pandas & NumPy Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 Pandas & NumPy Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「跨数据块增量聚合」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 Pandas & NumPy Academy 课中编写并运行代码吗？
能。每节 Pandas & NumPy Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。