如何修复Python中DeltaLake Apache PyArrow时间戳转换错误？

使用带Python绑定的DeltaLake Rust库从Azure Gen 2存储账户查询数据时，转换为Pandas DataFrame或PyArrow表时触发以下错误：

pyarrow.lib.ArrowInvalid: Casting from timestamp[ns] to timestamp[us, tz=UTC] would lose data: -4852202631933722624

该问题仅在查询特定Delta文件时出现，已知相关修复应该存在，但未明确根因。曾尝试手动修复损坏时间戳，但该方法速度过慢无法实用：

dl = DeltaTable(...)

min_time = pd.Timestamp("2020-01-01")
max_time = pd.Timestamp("2025-01-01")
schema = dl.schema().to_pyarrow()
cols = []
for field in schema:
    if isinstance(field.type, pa.TimestampType):
        cols.append(field.name)

for col in cols:
    predicate_min = f"{col} < '{str(min_time)}'"
    new_values = {col: str(min_time)}
    dl.update(predicate=predicate_min, new_values=new_values)
    predicate_max = f"{col} > '{str(max_time)}'"
    new_values = {col: str(max_time)}
    dl.update(predicate=predicate_max, new_values=new_values)

return dl.to_pandas()

当前困境：转换为PyArrow Table或Pandas DataFrame时因类型转换报错无法操作；转换为PyArrow Dataset可成功，但后续操作受限（如创建新架构）。需要找到高效修复错误或替换损坏时间戳的方法。

方法1：读取时直接修正时间戳（无需修改源数据） #

利用PyArrow Dataset读取后，通过compute模块批量修正超出范围的时间戳，再转换为Table/DataFrame，避免全量更新源数据：

import pyarrow as pa
import pyarrow.compute as pc
import pandas as pd
from deltalake import DeltaTable

dl = DeltaTable(...)
dataset = dl.to_pyarrow_dataset()

# 获取所有时间戳列
schema = dl.schema().to_pyarrow()
timestamp_cols = [f.name for f in schema if isinstance(f.type, pa.TimestampType)]

# 定义时间范围边界（纳秒级）
min_ts_ns = pd.Timestamp("2020-01-01").value
max_ts_ns = pd.Timestamp("2025-01-01").value

# 分批次处理数据
batches = []
for batch in dataset.to_batches():
    modified_batch = batch
    for col in timestamp_cols:
        col_data = modified_batch.column(col)
        # 替换小于最小值的时间戳
        corrected = pc.if_else(
            pc.less(col_data, pa.scalar(min_ts_ns, type=pa.timestamp('ns'))),
            pa.scalar(min_ts_ns, type=pa.timestamp('ns')),
            col_data
        )
        # 替换大于最大值的时间戳
        corrected = pc.if_else(
            pc.greater(corrected, pa.scalar(max_ts_ns, type=pa.timestamp('ns'))),
            pa.scalar(max_ts_ns, type=pa.timestamp('ns')),
            corrected
        )
        # 更新批次中的列
        modified_batch = modified_batch.set_column(
            modified_batch.schema.get_field_index(col),
            col,
            corrected
        )
    batches.append(modified_batch)

# 转换为PyArrow Table后转Pandas DataFrame
corrected_table = pa.Table.from_batches(batches)
df = corrected_table.to_pandas()

方法2：修改读取时的类型转换规则 #

强制将时间戳列读取为timestamp[ns]类型，规避PyArrow自动转换为微秒精度导致的报错：

from deltalake import DeltaTable
import pyarrow as pa

dl = DeltaTable(...)
original_schema = dl.schema().to_pyarrow()

# 修改所有时间戳列的精度为ns，移除时区约束
modified_schema = pa.schema(
    [
        pa.field(f.name, pa.timestamp('ns')) if isinstance(f.type, pa.TimestampType) else f
        for f in original_schema
    ]
)

# 使用修改后的schema读取数据
table = dl.to_pyarrow_table(schema=modified_schema)
df = table.to_pandas()

方法3：批量修复源数据（高效版） #

如果必须修复源数据，避免逐列逐条件的低效更新，改用批量读取-修正-覆盖写入的方式：

from deltalake import DeltaTable
import pyarrow as pa
import pyarrow.compute as pc
import pandas as pd

dl = DeltaTable(...)
# 读取全量数据为PyArrow Table
table = dl.to_pyarrow_dataset().to_table()

timestamp_cols = [f.name for f in table.schema if isinstance(f.type, pa.TimestampType)]
min_time = pd.Timestamp("2020-01-01")
max_time = pd.Timestamp("2025-01-01")

# 批量修正所有时间戳列
for col in timestamp_cols:
    table = table.set_column(
        table.schema.get_field_index(col),
        col,
        pc.clip(table.column(col), min_time, max_time)
    )

# 覆盖写入Delta表（注意：需确认数据可覆盖，或使用merge逻辑保留有效数据）
dl.write(table, mode="overwrite")

# 修复后即可正常转换
df = dl.to_pandas()

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