PyArrow Dataset转Pandas DataFrame空值问题：Bug还是超时？

核心问题分析 #

你遇到的是PyArrow Dataset在持续读取实时更新的Parquet文件时，超过设定的时间窗口后返回空DataFrame，但交互式环境下正常的问题。这大概率和Dataset的元数据缓存机制有关，而非线程池超时。

具体排查与修复方案 #

1. 禁用Dataset元数据缓存 #

PyArrow Dataset默认会缓存目录和文件的元数据，当外部进程持续更新文件时，缓存的元数据不会自动刷新，导致超过初始窗口后，新生成的文件无法被检测到。

• 解决方式：每次读取前重新创建Dataset，或配置文件系统自动刷新元数据：

import pyarrow.dataset as ds
import pyarrow.parquet as pq

# 方案1：每次读取前重新初始化Dataset
dataset = ds.dataset("/path/to/parquet_dir", format="parquet", use_legacy_dataset=False)

# 方案2：配置文件系统自动刷新
fs = pq.ParquetFileSystem("/path/to/parquet_dir", refresh=True)
dataset = ds.dataset("/path/to/parquet_dir", format="parquet", filesystem=fs)

2. 确保时间过滤逻辑动态计算 #

如果你的过滤条件是基于"当前时间-N分钟"，必须每次读取时重新计算时间范围，避免使用初始化时的静态值：

import pandas as pd

def get_recent_data(n_minutes=20):
    # 每次读取时重新计算时间窗口（确保时区和Parquet文件一致）
    end_time = pd.Timestamp.now(tz='UTC')
    start_time = end_time - pd.Timedelta(minutes=n_minutes)
    
    dataset = ds.dataset("/path/to/parquet_dir", format="parquet", use_legacy_dataset=False)
    table = dataset.to_table(
        filter=(ds.field("timestamp") >= start_time) & (ds.field("timestamp") <= end_time)
    )
    return table.to_pandas()

3. 检查文件分区识别是否正确 #

如果Parquet文件按时间分区（如year=2024/month=05），需确保Dataset能正确识别分区规则，否则过滤条件无法关联到分区，导致漏读新文件：

# 指定Hive风格分区模式
dataset = ds.dataset(
    "/path/to/parquet_dir", 
    format="parquet", 
    partitioning="hive", 
    use_legacy_dataset=False
)

4. 验证新文件的可见性 #

虽然已确保读写无竞争，但可通过手动遍历文件验证新生成的Parquet是否被正确写入：

import os
from pathlib import Path

def list_recent_files(dir_path, n_minutes=20):
    cutoff_time = pd.Timestamp.now() - pd.Timedelta(minutes=n_minutes)
    valid_files = []
    for path in Path(dir_path).rglob("*.parquet"):
        mtime = pd.Timestamp.fromtimestamp(os.path.getmtime(path))
        if mtime >= cutoff_time:
            valid_files.append(path)
    return valid_files

如果该函数能找到新文件但Dataset无法读取，可确认是元数据缓存问题。

5. 绕过Dataset直接读取文件 #

若缓存问题无法解决，可直接用PyArrow FileSystem遍历文件并过滤：

def read_recent_parquet(dir_path, n_minutes=20):
    end_time = pd.Timestamp.now(tz='UTC')
    start_time = end_time - pd.Timedelta(minutes=n_minutes)
    
    fs = pq.ParquetFileSystem(dir_path)
    all_files = fs.ls(dir_path, recursive=True)
    
    # 过滤出更新时间在窗口内的文件
    valid_files = [
        f for f in all_files 
        if fs.get_file_info(f).mtime >= start_time.timestamp()
    ]
    
    tables = []
    for f in valid_files:
        table = pq.read_table(
            f, 
            filters=[('timestamp', '>=', start_time), ('timestamp', '<=', end_time)]
        )
        if not table.empty:
            tables.append(table)
    
    return ds.concat_tables(tables).to_pandas() if tables else pd.DataFrame()

交互式环境正常的原因 #

交互式环境中，每次执行代码都会重新创建Dataset，元数据会重新加载；而长期运行的进程中，Dataset对象被复用，元数据缓存不会自动刷新，导致无法识别新生成的文件。

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