在Foundry中解析PDF原始数据集遇报错，如何生成目标结构数据集？

需要将包含多个PDF的原始数据集转换为Foundry数据集，要求数据集包含两列：doc_name（文档名称）和text（PDF文本内容），预期结构如下：

doc_name | text |
-----------------
xyz.pdf  | asdf |
yyz.pdf  | zxcv |

尝试使用pdfplumber等库处理时遇到两种报错：

1. 通过Foundry文件系统流打开文件时，报错io.UnsupportedOperation: File or stream is not seekable，对应代码：

with source.filesystem().open('xyz.pdf','rb') as f:
    pdf = pdfplumber.open(f)
    print(pdf)

2. 直接通过文件名打开时，报错FileNotFoundError: [Errno 2] No such file or directory: 'xyz.pdf'，对应代码：

pdf = pdfplumber.open('xyz.pdf')

解决方案 #

核心问题 #

Foundry提供的文件流是**不可随机读取（non-seekable）**的，而pdfplumber依赖的pdfminer组件需要可seek的流才能正常工作；直接用文件名找不到文件，是因为代码运行环境无法直接访问原始数据集的本地文件路径，必须通过Foundry的官方API读取文件内容。

解决步骤及代码 #

方案1：适配pdfplumber（转换为可seek流）

将Foundry返回的二进制内容转换为BytesIO对象（可seek的内存流），再传给pdfplumber处理，同时遍历所有PDF文件批量解析：

from pyspark.sql import SparkSession
import pdfplumber
from io import BytesIO
import pandas as pd

# 初始化Spark会话（Foundry环境中一般已自动配置）
spark = SparkSession.builder.getOrCreate()

# 读取原始数据集的二进制文件记录
source_rdd = spark.read.format("binaryFile").load("/path/to/your/source/dataset").rdd

def parse_single_pdf(file_data):
    # 提取文件路径和二进制内容
    file_path = file_data["path"]
    binary_content = file_data["content"]
    
    # 从路径中提取文档名称
    doc_name = file_path.split("/")[-1]
    
    # 将二进制内容转为可seek的BytesIO流
    with BytesIO(binary_content) as seekable_stream:
        with pdfplumber.open(seekable_stream) as pdf:
            # 提取所有页面文本并拼接
            full_text = "\n".join([page.extract_text() or "" for page in pdf.pages])
    
    return (doc_name, full_text)

# 批量解析所有PDF
parsed_results = source_rdd.map(parse_single_pdf).collect()

# 转换为DataFrame并写入目标Foundry数据集
result_df = pd.DataFrame(parsed_results, columns=["doc_name", "text"])
spark.createDataFrame(result_df).write.mode("overwrite").format("delta").save("/path/to/your/target/dataset")

方案2：改用Apache Tika（更适配非seek流）

如果pdfplumber仍有兼容性问题，可改用Tika，它对不可seek流的支持更好：

from pyspark.sql import SparkSession
from tika import parser
import pandas as pd

spark = SparkSession.builder.getOrCreate()
source_rdd = spark.read.format("binaryFile").load("/path/to/your/source/dataset").rdd

def parse_with_tika(file_data):
    file_path = file_data["path"]
    binary_content = file_data["content"]
    doc_name = file_path.split("/")[-1]
    
    # 直接传入二进制内容给Tika解析
    parsed_content = parser.from_buffer(binary_content)
    full_text = parsed_content.get("content", "").strip()
    
    return (doc_name, full_text)

parsed_results = source_rdd.map(parse_with_tika).collect()
result_df = pd.DataFrame(parsed_results, columns=["doc_name", "text"])
spark.createDataFrame(result_df).write.mode("overwrite").format("delta").save("/path/to/your/target/dataset")

注意事项 #

• 替换代码中的/path/to/your/source/dataset和/path/to/your/target/dataset为实际的Foundry数据集路径。
• 如果PDF中有扫描件（图片格式文本），需要额外配置OCR组件（如Tesseract）来提取文本。

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