如何用DataFlow读取需跳行和页脚的GCS Excel文件并写入BigQuery？

需求：将Google Cloud Storage（GCS）中的大量Excel文件读取并写入BigQuery，每个文件需设置skiprows=8和skipfooter=2参数。计划使用DataFlow，但不清楚实现方法。已尝试的代码使用ReadFromText处理Excel文件，存在逻辑问题。

原尝试代码：

if __name__ == '__main__':
   parser = argparse.ArgumentParser(description='Read all the startups')
   parser.add_argument('--output_prefix', default='/tmp/output', help='Output prefix')
   parser.add_argument('--input', default='gs://{folder_of_input}', help='Input directory')

   options, pipeline_args = parser.parse_known_args()
   p = beam.Pipeline(argv=pipeline_args)

   input = '{0}*.xlsx'.format(options.input)
   output_prefix = options.output_prefix

   # find most used packages
   (p
      | 'getExcel' >> beam.io.ReadFromText(input, skiprows=8, skipfooter=2)
      | 'write' >> beam.io.WriteToText(output_prefix)
   )

   p.run().wait_until_finish()

原代码核心问题 #

beam.io.ReadFromText仅适用于纯文本文件（如CSV、TXT），Excel是二进制格式文件，用该方法读取会得到乱码或无效内容，完全无法解析表格数据。必须使用专门的Excel解析库结合DataFlow的并行处理能力实现需求。

正确实现步骤 #

1. 安装依赖包
   本地开发和DataFlow Worker环境都需要安装以下依赖：

   pip install apache-beam pandas openpyxl google-cloud-bigquery fsspec gcsfs
   

2. 自定义Excel读取DoFn
   创建DoFn类，负责读取单个Excel文件、应用行跳过规则，并将数据转换为BigQuery可接受的字典格式。

3. 构建DataFlow Pipeline

   • 匹配GCS中的所有Excel文件
   • 并行处理每个文件
   • 将处理后的数据写入BigQuery

完整代码实现 #

import argparse
import pandas as pd
import apache_beam as beam
from apache_beam.io import fileio
from apache_beam.options.pipeline_options import PipelineOptions

class ReadExcelWithSkip(beam.DoFn):
    def process(self, file_metadata):
        # 获取GCS文件路径
        gcs_file_path = file_metadata.path
        
        # 读取Excel文件，跳过前8行和后2行
        df = pd.read_excel(
            gcs_file_path,
            skiprows=8,
            skipfooter=2,
            engine='openpyxl',
            dtype=str  # 统一转为字符串避免类型冲突，可按需调整
        )
        
        # 移除全空行（可选，根据实际数据情况添加）
        df = df.dropna(how='all')
        
        # 将每行数据转为字典，输出到下一个处理步骤
        for _, row in df.iterrows():
            yield row.to_dict()

def run():
    parser = argparse.ArgumentParser(description='Import Excel files from GCS to BigQuery')
    parser.add_argument('--input', required=True, help='GCS输入路径（带通配符），例如：gs://your-bucket/folder/*.xlsx')
    parser.add_argument('--bq_table', required=True, help='BigQuery目标表，格式：project.dataset.table')
    parser.add_argument('--temp_location', required=True, help='BigQuery写入用的GCS临时目录')
    args, pipeline_args = parser.parse_known_args()
    
    # 配置Pipeline参数
    pipeline_options = PipelineOptions(
        pipeline_args,
        temp_location=args.temp_location,
        project=args.bq_table.split('.')[0],  # 从表名提取项目ID
        region='us-central1'  # 根据实际区域调整
    )
    
    with beam.Pipeline(options=pipeline_options) as p:
        (
            p
            # 匹配GCS中的所有Excel文件
            | 'Match Excel Files' >> fileio.MatchFiles(args.input)
            # 获取文件可读元数据
            | 'Get File Metadata' >> fileio.ReadableFile()
            # 并行读取并处理每个Excel文件
            | 'Read & Process Excel' >> beam.ParDo(ReadExcelWithSkip())
            # 写入BigQuery
            | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
                args.bq_table,
                # 自动推断表结构，生产环境建议手动定义schema
                schema='SCHEMA_AUTODETECT',
                write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,  # 可选WRITE_TRUNCATE覆盖表数据
                create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
            )
        )

if __name__ == '__main__':
    run()

关键注意事项 #

• Schema管理：如果Excel结构固定，建议手动定义BigQuery schema（避免自动推断出错），可替换SCHEMA_AUTODETECT为beam.io.BigQuerySchema对象。
• Worker环境配置：使用DataFlow托管运行时，需通过--requirements_file参数指定依赖清单，或使用自定义Docker镜像确保依赖安装。
• 性能优化：超大型Excel文件可添加分块读取逻辑；调整Worker数量和机器类型提升并行处理效率。
• 权限配置：确保DataFlow服务账号拥有GCS读取权限和BigQuery写入权限。

内容的提问来源于stack exchange，提问作者le Minh Nguyen