从BigQuery批量生成2000个Matplotlib时间序列图的最优方案

核心结论 #

RowIterator.to_dataframe_iterable完全可行，是解决你问题的理想方案——它会将BigQuery查询结果分批转换成小DataFrame返回，不会一次性加载全部10GB数据到内存，完美避免内存溢出问题。结合查询排序和分批分组处理，能高效完成2000个序列的绘图任务，无需执行2000次单序列查询。

具体实现步骤 #

1. 优化BigQuery查询

先执行一次全局查询，按series_id和date排序，确保同一序列的时间数据连续出现，这样后续分批处理时可以避免跨批次缓存数据。

from google.cloud import bigquery
import matplotlib.pyplot as plt
import pandas as pd

# 初始化BigQuery客户端
client = bigquery.Client()

# 排序查询，确保同序列数据连续
query = """
SELECT series_id, date, value
FROM `your-project.your-dataset.your-table`
ORDER BY series_id, date
"""

# 使用to_dataframe_iterable分批获取数据，max_results可根据内存调整（默认10000行/批）
df_iter = client.query(query).result().to_dataframe_iterable(max_results=20000)

2. 分批处理并绘图

维护一个当前序列的缓存，每处理一批数据时，按series_id分组：遇到新序列时，先将缓存的上一个序列数据绘图并释放内存，再开始缓存新序列的数据。

current_series_id = None
current_data = []

for df_batch in df_iter:
    # 遍历当前批次内的所有序列分组
    for series_id, group in df_batch.groupby('series_id'):
        if series_id != current_series_id:
            # 处理上一个序列（如果存在）
            if current_series_id is not None:
                # 转换为结构化DataFrame
                df_series = pd.DataFrame(current_data, columns=['date', 'value'])
                df_series['date'] = pd.to_datetime(df_series['date'])
                
                # 绘图并保存
                plt.figure(figsize=(10, 6))
                plt.plot(df_series['date'], df_series['value'])
                plt.title(f'Time Series: {current_series_id}')
                plt.xlabel('Date')
                plt.ylabel('Value')
                plt.xticks(rotation=45)
                plt.tight_layout()
                plt.savefig(f'timeseries_{current_series_id}.png')
                plt.close()  # 关闭画布释放内存
                
                # 清空缓存
                current_data = []
            
            current_series_id = series_id
        
        # 将当前组的日期和值添加到缓存
        current_data.extend(group[['date', 'value']].values.tolist())

# 处理最后一个未完成的序列
if current_series_id is not None:
    df_series = pd.DataFrame(current_data, columns=['date', 'value'])
    df_series['date'] = pd.to_datetime(df_series['date'])
    
    plt.figure(figsize=(10, 6))
    plt.plot(df_series['date'], df_series['value'])
    plt.title(f'Time Series: {current_series_id}')
    plt.xlabel('Date')
    plt.ylabel('Value')
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig(f'timeseries_{current_series_id}.png')
    plt.close()

额外优化方案 #

方案一：提前过滤无效数据

在BigQuery查询中先过滤掉无意义的数据，减少返回的数据量：

SELECT series_id, date, value
FROM `your-project.your-dataset.your-table`
WHERE value IS NOT NULL  -- 过滤空值
  AND date BETWEEN '2020-01-01' AND '2023-12-31'  -- 限定时间范围
ORDER BY series_id, date

方案二：导出到分区文件处理

如果数据量极大，可先将数据按series_id分区导出到GCS，再逐个读取文件绘图：

1. BigQuery导出语句：

EXPORT DATA
OPTIONS(
    format = 'PARQUET',
    uri = 'gs://your-bucket/timeseries/*.parquet',
    partition_by = 'series_id'
) AS
SELECT series_id, date, value
FROM `your-project.your-dataset.your-table`

2. 读取分区文件绘图：

import glob
from google.cloud import storage

storage_client = storage.Client()
bucket = storage_client.bucket('your-bucket')
blobs = bucket.list_blobs(prefix='timeseries/')

for blob in blobs:
    if blob.name.endswith('.parquet'):
        # 直接读取GCS上的Parquet文件
        df = pd.read_parquet(f'gs://your-bucket/{blob.name}')
        df['date'] = pd.to_datetime(df['date'])
        
        # 解析文件名中的series_id
        series_id = blob.name.split('series_id=')[1].split('.')[0]
        
        plt.figure(figsize=(10, 6))
        plt.plot(df['date'], df['value'])
        plt.title(f'Time Series: {series_id}')
        plt.xlabel('Date')
        plt.ylabel('Value')
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.savefig(f'timeseries_{series_id}.png')
        plt.close()

内容的提问来源于stack exchange，提问作者Justin