Palantir Foundry提取XLSX文件报错，求解决方法及替代方案

一、错误原因与修复方法 #

1. 排查文件有效性 #

• 确认文件格式合法性：XLSX本质是zip压缩包，出现该错误大概率是文件本身问题——可能是上传不完整、文件损坏，或是后缀被修改的旧版XLS（二进制格式，非zip结构）。可以在Foundry中下载文件到本地，用解压工具尝试解压，或用Excel打开验证完整性。
• 过滤无效匹配文件：**/*.xlsx可能匹配到临时文件（如.~lock*.xlsx）或非目标文件，这类文件不是有效XLSX。打印filestatus的所有文件名，排除不符合要求的条目。

2. 代码层面修复 #

• 增加文件校验逻辑：在加载前先验证文件是否为合法zip包，避免直接抛出错误：

import zipfile

with source_df.filesystem().open(latest_file.path, 'rb') as f:
    try:
        with zipfile.ZipFile(f) as zf:
            if zf.testzip() is None:
                f.seek(0)  # 重置文件指针
                wb = openpyxl.load_workbook(f, read_only=True)
                # 后续解析逻辑
            else:
                raise ValueError("文件已损坏，无法作为XLSX读取")
    except zipfile.BadZipFile:
        raise ValueError("目标文件不是合法的XLSX格式")

• 修正循环缩进错误：原代码中rows.append(row_dict)缩进错误，只会添加最后一行数据，修正后：

rows = []
for row in ws.rows[2:]:
    row_dict = {}
    for i in range(len(headers)):
        row_dict[headers[i]] = row[i].value
    rows.append(row_dict)  # 将该行数据加入列表

二、Palantir Foundry中提取XLSX的其他可行方法 #

1. 使用Spark原生读取器 #

利用Spark内置的XLSX读取能力，无需手动处理文件系统，稳定性更高：

def compute(source_df, output_df, ctx):
    file_paths = [f"file://{f.path}" for f in source_df.filesystem().ls(glob='**/*.xlsx')]
    df = ctx.spark_session.read \
        .format("openpyxl") \
        .option("header", "true") \
        .option("dataAddress", "'Sheet1'!A2") \
        .load(file_paths)
    output_df.write_dataframe(df)

注意：需在代码库依赖中添加openpyxl包，版本匹配Foundry环境。

2. 使用Foundry Excel工具类 #

Foundry SDK提供了封装好的read_excel方法，简化Excel解析流程：

from transforms.api import transform, Input, Output
from transforms.excel import read_excel

@transform(
    output_df=Output("/path/to/output"),
    source_df=Input("/path/to/source")
)
def compute(source_df, output_df):
    df = read_excel(
        source_df,
        sheet_name="Sheet1",
        header_row=1,
        data_start_row=2
    )
    output_df.write_dataframe(df)

该方法自动处理文件系统交互和Excel解析细节，减少手动代码出错概率。

3. 批量处理多个XLSX文件 #

若需处理多份文件，可结合循环与read_excel实现批量合并：

def compute(source_df, output_df, ctx):
    dfs = []
    for file_status in source_df.filesystem().ls(glob='**/*.xlsx'):
        df = read_excel(
            source_df,
            sheet_name="Sheet1",
            header_row=1,
            data_start_row=2,
            file_path=file_status.path
        )
        dfs.append(df)
    combined_df = ctx.spark_session.unionByName(*dfs)
    output_df.write_dataframe(combined_df)

内容的提问来源于stack exchange，提问作者Murali krishna Ps