如何在Python中加载大型时序文件以开展分析？

一、逐行读取（轻量无依赖） #

一次性用read()把整个文件拉进内存是大文件卡壳的核心原因，换成逐行读取，内存占用极低，速度也快：

import gzip

# 用rt模式以文本格式读取，避免手动转码
with gzip.open('data.csv.gz', 'rt', encoding='utf-8') as f:
    for line in f:
        # 这里写单条数据的处理逻辑，比如解析时间戳和指标值
        timestamp, value = line.strip().split(',')
        # 后续时序分析操作

二、Pandas分块读取（适配数据分析场景） #

如果要做时序预处理或分析，Pandas原生支持gzip和分块加载，不用额外依赖：

import pandas as pd

# chunksize设为适合你机器内存的行数，比如1万行/块
chunk_generator = pd.read_csv('data.csv.gz', compression='gzip', chunksize=10000)

for chunk in chunk_generator:
    # 转换时间列为时序格式
    chunk['timestamp'] = pd.to_datetime(chunk['timestamp'])
    # 对当前块做分析，比如计算分钟级均值
    chunk_res = chunk.resample('T', on='timestamp')['value'].mean()
    # 可以把结果存到文件或累加

三、Dask（单机器处理超大规模数据） #

数据量到GB甚至TB级，单机器用Dask更高效，语法和Pandas几乎一致，自动并行处理：

import dask.dataframe as dd

# 支持读取多个.gz文件，用通配符匹配
dask_df = dd.read_csv('*.csv.gz', compression='gzip', blocksize='64MB')
# 按时间分组计算统计量，compute()才会实际执行计算
hourly_mean = dask_df.groupby('timestamp').mean().resample('H').mean().compute()

四、Spark（分布式集群处理） #

如果单机器完全扛不住，上Spark分布式框架，自动把任务拆分到集群节点：

from pyspark.sql import SparkSession
from pyspark.sql.functions import window

spark = SparkSession.builder.appName("TimeSeriesGz").getOrCreate()
# 读取.gz文件，自动识别压缩格式
spark_df = spark.read.csv('data.csv.gz', header=True, inferSchema=True)
# 转换时间列类型
spark_df = spark_df.withColumn('timestamp', spark_df['timestamp'].cast('timestamp'))
# 滑动窗口统计，比如每小时的指标计数
window_stats = spark_df.groupBy(window('timestamp', '1 hour')).count()
window_stats.show()

内容的提问来源于stack exchange，提问作者Fernando Victoria Valpuesta