Databricks Spark中用pypdf读取多PDF遇KeyError: '/Root'求助

可能原因及对应解决方案 #

1. 定位损坏的PDF文件 #

报错KeyError: '/Root'多数情况是某份PDF本身结构异常或损坏，单条测试时未覆盖到问题文件，多条数据时触发了错误。可以在UDF中添加异常捕获，标记错误记录来定位问题文件：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
import pypdf

def extract_pdf_content(pdf_bytes):
    try:
        reader = pypdf.PdfReader(pdf_bytes)
        return "\n".join([page.extract_text() for page in reader.pages])
    except Exception as e:
        return f"ERROR: {str(e)}"

extract_pdf_udf = udf(extract_pdf_content, StringType())

# 筛选出报错的记录
error_records = df.withColumn("content", extract_pdf_udf(df.pdf_content)) \
                  .filter("content LIKE 'ERROR:%'")
display(error_records)

2. 避免UDF内重用pypdf实例 #

如果UDF外部初始化了全局的PdfReader实例，Spark并行处理时多任务共享同一实例会导致状态混乱，引发异常。必须在UDF内部每次调用都创建新的实例：

# 错误写法：全局实例会导致多任务冲突
# reader = pypdf.PdfReader()

# 正确写法：每次调用UDF都新建实例
def extract_pdf_content(pdf_bytes):
    reader = pypdf.PdfReader(pdf_bytes)
    return "\n".join([page.extract_text() for page in reader.pages])

3. 验证PDF字节流完整性 #

确保传入UDF的PDF字节流没有截断或编码错误，可尝试将字节流写入临时文件再读取（排查字节流问题）：

import tempfile
import os

def extract_pdf_content(pdf_bytes):
    try:
        with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp_file:
            tmp_file.write(pdf_bytes)
            tmp_path = tmp_file.name
        reader = pypdf.PdfReader(tmp_path)
        content = "\n".join([page.extract_text() for page in reader.pages])
        os.unlink(tmp_path)
        return content
    except Exception as e:
        return f"ERROR: {str(e)}"

4. 升级pypdf版本 #

旧版本pypdf对部分PDF结构兼容性差，可能在多文件处理时触发bug。在Databricks Notebook中执行以下命令升级：

%pip install --upgrade pypdf

UDF调试技巧 #

用mapPartitions替代UDF在小数据集上测试，打印每条记录的处理日志，方便定位问题：

def debug_pdf_processing(partition):
    for row in partition:
        pdf_bytes = row.pdf_content
        try:
            reader = pypdf.PdfReader(pdf_bytes)
            print(f"处理成功，元数据：{row.metadata}")
            yield (row.id, "\n".join([page.extract_text() for page in reader.pages]))
        except Exception as e:
            print(f"处理失败，记录ID：{row.id}，错误：{str(e)}")
            yield (row.id, f"ERROR: {str(e)}")

# 取5条样本测试
sample_df = df.limit(5)
debug_result = sample_df.rdd.mapPartitions(debug_pdf_processing).toDF(["id", "content"])
display(debug_result)

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