Python中传感器数据存储优化：按X兆字节分割的最佳方法

一、更优实现方式与流式压缩方案 #

1. 写入性能优化 #

当前逐行转换字符串并写入的方式IO开销极高，可通过以下方式优化：

• 批量写入：攒一批数据（比如10秒共640条）再一次性写入磁盘，大幅减少IO调用次数。磁盘IO是高耗时操作，批量写入能显著降低处理延迟。
• 减少字符串转换开销：提前格式化每行内容，用列表批量存储行字符串，最后通过join合并成大字符串再编码写入，比循环拼接字符串效率更高。

2. 流式压缩方案 #

流式压缩完全适配实时处理场景，无需先存原始文件再压缩，推荐以下两种格式：

• Gzip：Python标准库自带gzip模块，支持流式写入，压缩速度快，文本数据压缩率可达1/3_{1/5，150MB原始CSV可压缩至30}50MB，对实时处理的性能影响极小。
• Zstd：压缩率比Gzip更高，读写速度也不逊色，需通过zstandard第三方库实现，适合对存储占用要求更严格的场景。

示例代码（Gzip批量写入）

import gzip
from datetime import datetime

# 以追加模式打开gzip文件
with gzip.GzipFile("sensor_data.gz", "ab") as log_file:
    batch = []
    # 模拟每秒64条数据，攒10秒的量
    for _ in range(640):
        datet = datetime.now()
        milivolts = 123.45  # 替换为实际传感器数值
        line = f"{datet.isoformat()},{milivolts}\n"
        batch.append(line)
    # 批量写入，先合并字符串再编码
    log_file.write("".join(batch).encode("utf-8"))

二、按指定大小分割文件（无数据间隙） #

要实现按X兆字节分割文件且不截断数据，核心是跟踪当前文件大小，达到阈值时切换新文件，关键细节如下：

1. 设定大小阈值：将X兆转换为字节数（如50MB = 50 * 1024 * 1024字节）。
2. 实时检查文件大小：每次批量写入前，检查当前压缩文件的实际大小，若写入后会超过阈值，则关闭当前文件，生成新的序号命名文件（如sensor_001.gz）。
3. 处理剩余数据：程序退出时确保剩余未批量的数据全部写入，避免丢失。

示例代码（结合Gzip与文件分割）

import gzip
import os
from datetime import datetime

MAX_FILE_SIZE = 50 * 1024 * 1024  # 50MB，可自定义
current_file_idx = 1
current_log_file = None

def create_new_log_file():
    global current_file_idx
    filename = f"sensor_data_{current_file_idx:03d}.gz"
    current_file_idx += 1
    return gzip.GzipFile(filename, "wb")

# 初始化第一个文件
current_log_file = create_new_log_file()
batch = []

try:
    # 模拟持续采集传感器数据
    while True:
        datet = datetime.now()
        milivolts = 123.45  # 替换为实际传感器数值
        line = f"{datet.isoformat()},{milivolts}\n"
        batch.append(line)
        
        # 每攒10秒的数据写入一次
        if len(batch) >= 640:
            batch_data = "".join(batch).encode("utf-8")
            # 检查当前文件写入后是否超出阈值
            current_size = os.path.getsize(current_log_file.name) if os.path.exists(current_log_file.name) else 0
            if current_size + len(batch_data) > MAX_FILE_SIZE:
                current_log_file.close()
                current_log_file = create_new_log_file()
            current_log_file.write(batch_data)
            batch = []
finally:
    # 写入剩余未批量的数据
    if batch:
        current_log_file.write("".join(batch).encode("utf-8"))
    current_log_file.close()

注意事项 #

• 压缩文件的实际大小与原始数据大小并非线性对应，通过os.path.getsize()直接检查压缩文件的大小是最准确的方式。
• 文件名使用序号或时间戳命名，方便后续数据的排序与检索。

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