你提到的痛点非常典型——Feather的速度和类型保留优势很吸引人，但不支持行追加确实是个问题。结合你的需求（故障安全、避免全量重写、二进制格式），这里有几个经过实践验证的方案，按推荐优先级排序：

方案1：分块Feather文件+事后合并（最贴合你的偏好） #

核心思路是将数据按固定量或时间分片，每次写入独立的Feather小文件，故障时已生成的文件全部保留，最终通过合并得到完整数据集。完全使用Feather格式，兼顾速度、类型安全和故障韧性。

实现代码示例 #

import pandas as pd
from datetime import datetime
import os
import time

# 配置参数：每积累1000行（约50秒数据）写入一个文件
CHUNK_SIZE = 1000
STORAGE_DIR = "feather_data_chunks"
os.makedirs(STORAGE_DIR, exist_ok=True)

# 临时缓存数据，避免频繁写入文件
data_buffer = []

try:
    while True:
        # 模拟生成20行设备数据（替换成你的实际数据生成逻辑）
        new_data = pd.DataFrame({
            "timestamp": [datetime.now()] * 20,
            "sensor_temp": [25.0 + i*0.01 for i in range(20)],
            "sensor_humidity": [60.0 + i*0.05 for i in range(20)],
            "device_status": ["normal"] * 20
        })
        data_buffer.append(new_data)
        
        # 检查缓存是否达到阈值，达到则写入文件
        total_rows = sum(df.shape[0] for df in data_buffer)
        if total_rows >= CHUNK_SIZE:
            chunk_df = pd.concat(data_buffer, ignore_index=True)
            # 用时间戳命名文件，确保唯一性
            file_name = f"chunk_{datetime.now().strftime('%Y%m%d_%H%M%S')}.feather"
            chunk_df.to_feather(os.path.join(STORAGE_DIR, file_name))
            # 清空缓存
            data_buffer = []
        
        # 模拟每秒生成一次数据
        time.sleep(1)

except Exception as e:
    print(f"程序异常终止: {str(e)}")
    # 故障时将剩余缓存写入文件，避免丢失
    if data_buffer:
        chunk_df = pd.concat(data_buffer, ignore_index=True)
        file_name = f"chunk_{datetime.now().strftime('%Y%m%d_%H%M%S')}_failed.feather"
        chunk_df.to_feather(os.path.join(STORAGE_DIR, file_name))
    print("未写入的剩余数据已保存")

# 事后合并所有分块文件的函数
def merge_feather_chunks(input_dir, output_path):
    all_feather_files = [os.path.join(input_dir, f) for f in os.listdir(input_dir) if f.endswith(".feather")]
    full_df = pd.concat([pd.read_feather(file) for file in all_feather_files], ignore_index=True)
    full_df.to_feather(output_path)
    return full_df

# 调用合并（程序正常结束后执行）
# merge_feather_chunks(STORAGE_DIR, "full_device_data.feather")

方案优势 #

• 完全基于Feather，读写速度拉满，数据类型100%保留
• 故障安全：每个分块文件写入后立即持久化，不会因为单次故障丢失之前的所有数据
• 存储高效：Feather的压缩率远高于CSV
• 实现简单：无需引入额外依赖，逻辑清晰

小缺点 #

• 最终需要合并多个文件，但Feather的读取速度极快，合并168个小时级分块（一周数据）的开销几乎可以忽略

方案2：HDF5追加写入+最终转Feather #

如果偏好单文件管理，可以用HDF5作为中间存储（原生支持追加），程序正常结束后再转换为Feather格式。HDF5同样是二进制格式，保留数据类型，且支持增量写入。

实现代码示例 #

import pandas as pd
from datetime import datetime
import time

# 配置参数
HDF_FILE = "device_data_temp.h5"
DATA_KEY = "sensor_records"

try:
    # 以追加模式打开HDF存储
    with pd.HDFStore(HDF_FILE, mode="a") as store:
        while True:
            # 模拟生成20行数据
            new_data = pd.DataFrame({
                "timestamp": [datetime.now()] * 20,
                "sensor_temp": [25.0 + i*0.01 for i in range(20)],
                "sensor_humidity": [60.0 + i*0.05 for i in range(20)],
                "device_status": ["normal"] * 20
            })
            # 追加写入HDF，需指定format='table'支持追加
            store.append(DATA_KEY, new_data, format="table", data_columns=True)
            
            time.sleep(1)

except Exception as e:
    print(f"程序异常终止: {str(e)}")
    print("HDF文件中已完整保存所有生成的数据")

# 正常结束后转换为Feather格式
def hdf_to_feather(hdf_path, data_key, output_path):
    with pd.HDFStore(hdf_path, mode="r") as store:
        full_df = store[data_key]
    full_df.to_feather(output_path)
    return full_df

# 调用转换
# hdf_to_feather(HDF_FILE, DATA_KEY, "full_device_data.feather")

方案优势 #

• 单文件管理更方便，无需处理多个分块
• 原生支持追加，无需手动缓存数据
• 故障时数据自动持久化，不会丢失

小缺点 #

• HDF5的读写速度略逊于Feather
• 格式通用性不如Feather（部分数据分析工具对Feather支持更好）

方案3：Dask + Feather增量写入（适合超大型数据集） #

如果你的数据量未来会进一步增长，Dask是个不错的选择——它支持Feather的分块增量写入，同时能处理远超内存的大型数据集。

实现代码示例 #

import dask.dataframe as dd
import pandas as pd
from datetime import datetime
import os
import time

STORAGE_DIR = "dask_feather_storage"
os.makedirs(STORAGE_DIR, exist_ok=True)

try:
    while True:
        # 模拟生成20行数据
        new_data = pd.DataFrame({
            "timestamp": [datetime.now()] * 20,
            "sensor_temp": [25.0 + i*0.01 for i in range(20)],
            "sensor_humidity": [60.0 + i*0.05 for i in range(20)],
            "device_status": ["normal"] * 20
        })
        # 转换为Dask DataFrame并追加写入
        ddf = dd.from_pandas(new_data, npartitions=1)
        ddf.to_feather(STORAGE_DIR, append=True)
        
        time.sleep(1)

except Exception as e:
    print(f"程序异常终止: {str(e)}")

# 读取完整数据并转换为单Feather文件
full_ddf = dd.read_feather(STORAGE_DIR)
full_df = full_ddf.compute()
full_df.to_feather("full_device_data.feather")

方案优势 #

• 原生支持Feather增量写入，无需手动分块
• 支持分布式处理，适合超大规模数据
• 故障安全，已写入的分块不会丢失

小缺点 #

• 需要引入Dask依赖，学习成本略高
• 对于你的当前规模（一周约1200万行），有点大材小用

总结推荐 #

优先选择方案1，完全贴合你对Feather的偏好，实现简单，故障安全，且性能和存储效率都能满足需求。如果偏好单文件管理，方案2是可靠的替代。

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