如何使用Python每日查询超大型在线CSV文件获取新冠疫苗接种新增统计数据

首次统计实现方案 #

• 优先选择S3 Select服务端查询
  该CSV存储在AWS S3上，你可以直接调用S3 Select接口，将你需要的分组统计SQL语句直接传给S3服务端，服务端完成计算后仅返回最终统计结果，无需下载任何多余的文件内容，是效率最高、资源消耗最低的方案。Python可通过boto3库调用该能力，参考代码如下：

import boto3

s3 = boto3.client('s3')
resp = s3.select_object_content(
    Bucket='对应S3存储桶名',
    Key='CSV文件的S3路径',
    ExpressionType='SQL',
    Expression="select city, day, count(*) from s3object group by city, day",
    InputSerialization={'CSV': {"FileHeaderInfo": "USE"}, 'CompressionType': 'NONE'},
    OutputSerialization={'CSV': {}},
)
# 处理返回的结果流写入本地存储即可

• 备选：流式读取本地计算
  如果你无法使用S3 Select，可通过s3fs或smart-open库直接打开远程S3文件，配合Python内置csv模块逐行读取，用collections.defaultdict(int)做分组计数，全程不需要加载整个文件到内存，也不用下载完整文件。

每日增量更新方案 #

该文件为追加写入模式，你可以通过以下逻辑实现仅处理新增内容：

1. 每次统计完成后，持久化记录两个信息：当前S3文件的Content-Length字节大小、上次读取末尾剩余的不完整行片段
2. 每日执行更新任务时，先仅请求S3文件的元数据，对比当前文件的Content-Length和你记录的历史字节值：
   • 数值无变化说明没有新增数据，直接结束任务
   • 数值变大说明有新增内容，使用HTTP Range头请求bytes=历史字节大小-1024:，多请求1KB是为了补全上次读取时可能被截断的不完整行
3. 将请求到的新增内容片段和上次留存的不完整行拼接后按行拆分，拆分后的最后一行如果不完整就留存到下次使用，前面的完整行解析后，对应累加你存储的统计结果中对应city和day的计数即可

结果存储建议 #

统计结果可使用轻量的SQLite存储，建表语句参考：

CREATE TABLE IF NOT EXISTS vaccine_stats (
    city TEXT,
    day DATE,
    count INT,
    PRIMARY KEY (city, day)
);

更新时直接用INSERT OR REPLACE语句累加对应分组的计数即可，后续查询统计结果也非常方便。

内容的提问来源于stack exchange，提问作者Luiz Fernando Puttow Southier