优化传感器数据磁盘写入：减少写入次数与文件体积

针对你的长期传感器数据采集需求，以下几个方案可以解决「减少SSD写入、保持高压缩率、保留历史数据」的问题：

1. 直接用Parquet追加写入（最简化修改） #

你之前误解了Parquet的追加逻辑——Parquet支持带压缩的追加写入，不需要每次全量重写历史数据。借助pyarrow或fastparquet引擎，pandas的to_parquet可以通过mode='append'参数实现只写入新数据，同时保留brotli压缩。

修改后的核心代码：

POLL_EVERY = 15.0  # seconds
DATANAME = 'sensor_data.parquet'
BATCH_SIZE = 1000  # 攒够1000条再写入，可根据需求调整
columns = ['timestamp', 'machine', 'load', 'freq', 'watts', 'temp']  # 替换为实际列

# 初始化批量缓存
batch_buffer = []

while True:
    for machine in machines:
        data = get_data(machine)
        batch_buffer.append(data)
    
    # 达到批量大小则写入
    if len(batch_buffer) >= BATCH_SIZE:
        new_df = pandas.DataFrame(batch_buffer, columns=columns)
        # 追加写入，保持brotli压缩
        new_df.to_parquet(DATANAME, mode='append', compression='brotli', index=False, engine='pyarrow')
        batch_buffer = []  # 清空缓存
    
    time.sleep(POLL_EVERY)

• 优势：不用修改现有存储格式，每次仅写入新的批量数据，完全避免全量重写，压缩率保持不变。
• 注意：首次运行需要确保文件存在（或先创建空文件），后续直接追加即可；如果担心数据丢失，可以每小时额外做一次增量备份。

2. 按时间分区存储Parquet文件（彻底避免重写） #

如果想完全杜绝对历史文件的修改，可按时间维度拆分Parquet文件，比如按天/小时生成独立文件：

核心实现逻辑：

POLL_EVERY = 15.0
BATCH_SIZE = 1000
columns = ['timestamp', 'machine', 'load', 'freq', 'watts', 'temp']
batch_buffer = []

while True:
    for machine in machines:
        data = get_data(machine)
        batch_buffer.append(data)
    
    if len(batch_buffer) >= BATCH_SIZE:
        new_df = pandas.DataFrame(batch_buffer, columns=columns)
        # 获取当前时间作为分区键（比如按小时）
        current_hour = pandas.Timestamp.now().strftime('%Y%m%d_%H')
        filename = f'sensor_data_{current_hour}.parquet'
        # 追加到对应小时的文件
        new_df.to_parquet(filename, mode='append', compression='brotli', index=False, engine='pyarrow')
        batch_buffer = []
    
    time.sleep(POLL_EVERY)

• 优势：历史文件一旦生成就不再修改，彻底保护SSD写入寿命；查询时可按时间过滤，只加载需要的文件，提升查询效率。
• 扩展：可以定期将多天的小文件合并为大文件（比如每周合并一次），进一步优化压缩率和存储效率，合并操作可在低峰期后台执行。

3. 混合暂存+批量合并（优化写入性能） #

如果追求极致的实时写入速度，可先用轻量格式暂存数据，再批量合并到压缩Parquet：

• 暂存层用Feather格式（比CSV快数倍，支持列式存储），每次采集直接追加到暂存文件；
• 当暂存文件达到阈值（比如1GB），在后台线程将其合并到历史压缩Parquet文件，合并时可重新优化压缩（批量数据压缩率更高）。

核心代码示例：

import threading
import pandas as pd
import os

POLL_EVERY = 15.0
STAGING_FILE = 'sensor_staging.feather'
HISTORY_FILE = 'sensor_history.parquet'
STAGING_SIZE_THRESHOLD = 1 * 1024 * 1024 * 1024  # 1GB
columns = ['timestamp', 'machine', 'load', 'freq', 'watts', 'temp']

def merge_staging_to_history():
    if not os.path.isfile(STAGING_FILE):
        return
    # 读取暂存数据
    staging_df = pd.read_feather(STAGING_FILE)
    # 追加到历史文件
    staging_df.to_parquet(HISTORY_FILE, mode='append', compression='brotli', index=False, engine='pyarrow')
    # 清空暂存文件
    open(STAGING_FILE, 'w').close()

batch_buffer = []
while True:
    for machine in machines:
        data = get_data(machine)
        batch_buffer.append(data)
    
    if len(batch_buffer) >= 100:
        new_df = pd.DataFrame(batch_buffer, columns=columns)
        # 追加到暂存Feather文件
        new_df.to_feather(STAGING_FILE, mode='append')
        batch_buffer = []
    
    # 检查暂存文件大小，达到阈值则启动后台合并
    if os.path.isfile(STAGING_FILE) and os.path.getsize(STAGING_FILE) >= STAGING_SIZE_THRESHOLD:
        threading.Thread(target=merge_staging_to_history, daemon=True).start()
    
    time.sleep(POLL_EVERY)

• 优势：实时写入几乎无延迟，后台合并不影响采集流程；暂存文件写入速度快，合并时批量压缩效率更高。

额外优化建议 #

• 数据去重：因为你的数据有大量重复值，可在批量写入前执行new_df = new_df.drop_duplicates()，进一步减少写入数据量；
• 调整Parquet行组大小：通过row_group_size参数设置合适的行组（比如100万行），平衡压缩率和写入性能；
• 定期校验：每月对历史Parquet文件做一次完整性校验，避免数据损坏。

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