如何在Scrapy中将TSV文件转换为Parquet后上传至S3

要在上传S3前完成TSV转Parquet，核心是修改自定义的FilesPipeline，在文件下载完成后、上传到S3前执行转换逻辑，替换待上传的文件。具体步骤如下：

1. 确保依赖安装 #

首先确认pyarrow已安装（你本地转换逻辑依赖它）：

pip install pyarrow

同时确保Scrapy的S3存储依赖（如boto3）已就绪，保证能正常上传文件到S3。

2. 修改自定义FilesPipeline #

替换你现有的OriginalNameFilesPipeline，重写file_path和item_completed方法来实现转换：

import os
from datetime import datetime
from scrapy.pipelines.files import FilesPipeline
from itemadapter import ItemAdapter
import pyarrow.csv as csv
import pyarrow.parquet as pq

class OriginalNameFilesPipeline(FilesPipeline):
    def file_path(self, request, response=None, info=None):
        # 生成带日期的parquet格式文件名
        file_name_tsv = request.url.split("=")[-1]
        base_name = file_name_tsv.removesuffix('.tsv')
        file_name = f"{base_name}_{datetime.today().strftime('%Y%m%d')}.parquet"
        return file_name

    def item_completed(self, results, item, info):
        adapter = ItemAdapter(item)
        if not adapter.get('files'):
            return item

        # 处理每个下载完成的TSV文件
        for ok, file_info in results:
            if ok:
                local_tsv_path = file_info['path']
                # 构建Parquet文件的本地临时路径
                local_parquet_path = local_tsv_path.replace('.tsv', '.parquet')
                
                # 执行TSV转Parquet转换
                parse_options = csv.ParseOptions(delimiter="\t")
                table = csv.read_csv(local_tsv_path, parse_options=parse_options)
                pq.write_table(table, local_parquet_path)
                
                # 更新文件信息，让Scrapy上传Parquet文件而非原TSV
                file_info['path'] = local_parquet_path
                # 清理原TSV临时文件（可选，节省磁盘空间）
                if os.path.exists(local_tsv_path):
                    os.remove(local_tsv_path)
        
        return item

3. 关键逻辑说明 #

• file_path方法：直接生成.parquet后缀的文件名，确保最终上传到S3的文件是目标格式的命名。
• item_completed方法：这是FilesPipeline的下载完成钩子，在这里完成：
  1. 获取本地临时TSV文件的路径；
  2. 复用你本地的转换逻辑，将TSV转为Parquet；
  3. 替换文件信息中的路径为Parquet文件路径，让Scrapy后续上传这个文件到S3；
  4. 可选删除原TSV文件，清理临时存储。

4. 配置验证 #

确保settings.py中已启用该Pipeline，且FILES_STORE配置为你的S3存储路径（例如s3://your-bucket/folder/），同时S3的访问权限配置正确。

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