求助：使用Dataflow Runner无法将Datastore数据转PDF存GCS

核心问题点 #

DirectRunner本地运行时，代码直接操作本地文件系统，写完还能手动传GCS；但Dataflow的Worker是临时容器，本地文件会随任务结束销毁，分布式环境下直接写本地文件根本没法保证同步到GCS，另外PDF生成依赖的库可能没在Worker环境装。

具体解决步骤 #

1. 跳过本地文件，直接在内存生成PDF并传GCS #

别先写本地文件，用内存流生成PDF字节数据，直接上传GCS。比如用reportlab的话，先写入BytesIO，再用GCS客户端上传：

from reportlab.pdfgen import canvas
from google.cloud import storage
import io

def make_and_upload_pdf(data, gcs_bucket, gcs_path):
    # 内存里生成PDF
    buffer = io.BytesIO()
    c = canvas.Canvas(buffer)
    # 示例：把数据写入PDF
    c.drawString(100, 750, f"内容: {data}")
    c.showPage()
    c.save()
    buffer.seek(0)
    
    # 上传到GCS
    storage_client = storage.Client()
    bucket = storage_client.bucket(gcs_bucket)
    blob = bucket.blob(gcs_path)
    blob.upload_from_file(buffer, content_type='application/pdf')

2. 给Dataflow Worker装PDF依赖库 #

如果用了reportlab、fpdf这类库，得让Worker环境也装上：

• 写个requirements.txt，里面列上依赖：

  reportlab>=3.6.0
  google-cloud-storage>=2.0.0
  

  运行Dataflow时加参数：--requirements_file requirements.txt
• 或者用setup.py，内容如下，运行时加--setup_file setup.py：

  from setuptools import setup
  
  setup(
      name='pdf-gen-job',
      version='0.1',
      install_requires=[
          'reportlab>=3.6.0',
          'google-cloud-storage>=2.0.0'
      ],
  )
  

3. 调整Pipeline写法，适配分布式环境 #

用ParDo处理每个从Datastore读来的实体，直接调用上传逻辑，别依赖本地状态：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions

class GeneratePDF(beam.DoFn):
    def __init__(self, gcs_bucket):
        self.gcs_bucket = gcs_bucket
        
    def process(self, element):
        # element是Datastore实体
        data_id = element.key.id_or_name
        gcs_path = f"pdfs/{data_id}.pdf"
        make_and_upload_pdf(element, self.gcs_bucket, gcs_path)
        return [gcs_path]

def run():
    options = PipelineOptions()
    gcp_opts = options.view_as(GoogleCloudOptions)
    gcp_opts.project = "你的项目ID"
    gcp_opts.job_name = "datastore转pdf任务"
    gcp_opts.staging_location = "gs://你的桶/staging"
    gcp_opts.temp_location = "gs://你的桶/temp"
    gcp_opts.region = "us-central1"
    
    with beam.Pipeline(options=options) as p:
        (p
         | '读Datastore数据' >> beam.io.ReadFromDatastore(
             project="你的项目ID",
             query=beam.io.gcp.datastore.v1new.types.Query(kind="你的实体类型")
         )
         | '生成并上传PDF' >> beam.ParDo(GeneratePDF(gcs_bucket="你的桶"))
         | '打印GCS路径' >> beam.Map(print)
        )

if __name__ == '__main__':
    run()

4. 检查权限配置 #

确保Dataflow的服务账号有这些权限：

• Datastore读取权限（roles/datastore.viewer及以上）
• GCS写入权限（roles/storage.objectCreator及以上）
  去IAM控制台给Dataflow默认服务账号（[项目编号]-compute@developer.gserviceaccount.com）加对应角色就行。

验证方式 #

跑Dataflow任务后，直接看GCS目标路径有没有生成PDF，同时去Dataflow控制台看日志，排查有没有依赖缺失、权限不够这类报错。

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