Python3.8运行Apache Beam Dataflow报storage未定义问题咨询

• 你遇到的NameError: name 'storage' is not defined和Python版本、google-cloud-storage版本没有关系，核心原因是代码存在语法错误：导入smart_open的语句和类定义被粘在了同一行：

  # 错误写法，open和class关键字连写导致解析异常
  from smart_open import openclass WriteCSVFile(beam.DoFn):
  

  Python解析时会把openclass识别为要从smart_open导入的对象，直接导致代码解析逻辑异常，连带storage的导入识别失效。
• 你当前的实现逻辑本身不符合Apache Beam的运行设计：在DoFn中手动初始化GCS客户端、单线程上传文件属于反模式，在Dataflow分布式环境下还会触发客户端序列化失败、多worker写同一路径文件覆盖、性能不足等额外问题。

Apache Beam本身内置了GCS文件系统的读写实现，不需要额外依赖google-cloud-storage库，直接使用原生IO组件即可完成JSON转CSV写入GCS的需求，也是Dataflow官方推荐的标准实现方式，稳定性和性能远高于手动调用storage客户端的写法。

实现代码 #

import apache_beam as beam
import csv
import json
import io
from apache_beam.options.pipeline_options import PipelineOptions

# 按实际情况修改配置
GCP_PROJECT_ID = "你的GCP项目ID"
DATAFLOW_REGION = "us-central1" # 替换为你使用的区域
GCS_TEMP_PATH = "gs://你的存储桶/temp/"
INPUT_JSON_PATH = "gs://你的存储桶/input/*.json" # 支持通配符匹配多文件
OUTPUT_CSV_PATH = "gs://你的存储桶/output/output_poc"
CSV_COLUMNS = [
    'account_id', 'isActive', 'balance', 'age', 'eyeColor',
    'name', 'gender', 'company', 'email', 'phone', 'address'
]

def convert_json_to_csv_row(record: dict) -> str:
    """单条JSON记录转标准CSV行，自动处理特殊字符转义"""
    buffer = io.StringIO()
    writer = csv.writer(buffer, quoting=csv.QUOTE_MINIMAL)
    writer.writerow([str(record.get(col, "")) for col in CSV_COLUMNS])
    return buffer.getvalue().strip()

def run():
    pipeline_options = PipelineOptions(
        runner="DataflowRunner",
        project=GCP_PROJECT_ID,
        region=DATAFLOW_REGION,
        temp_location=GCS_TEMP_PATH,
    )

    with beam.Pipeline(options=pipeline_options) as p:
        # 读取JSON文件（支持直接读GCS路径，自动按行读取）
        json_records = (
            p
            | "ReadSourceFile" >> beam.io.ReadFromText(INPUT_JSON_PATH)
            | "ParseJSONContent" >> beam.Map(json.loads)
        )

        # 生成CSV表头
        csv_header = [",".join(CSV_COLUMNS)]
        # 转换所有数据行
        csv_data_rows = json_records | "ConvertToCSVFormat" >> beam.Map(convert_json_to_csv_row)

        # 合并表头和数据写入GCS
        (
            (csv_header, csv_data_rows)
            | "MergeHeaderAndData" >> beam.Flatten()
            | "WriteCSVToGCS" >> beam.io.WriteToText(
                OUTPUT_CSV_PATH,
                file_name_suffix=".csv",
                num_shards=1, # 小数据量设为1输出单个文件，大数据量删除该参数自动分片提升性能
                shard_name_template="" # 配合num_shards=1使用，输出文件名无分片后缀
            )
        )

if __name__ == "__main__":
    run()

方案说明 #

• 全程不需要导入google.cloud.storage，所有GCS读写逻辑由Beam内置的GCS客户端实现完成，不存在依赖版本冲突问题。
• 自动适配Dataflow分布式运行环境，内置IO重试、分片、性能优化逻辑，支持TB级数据处理，不会出现多worker文件冲突问题。
• 内置CSV格式转义逻辑，避免字段中包含逗号、换行符、引号导致的CSV格式错乱问题。
• 如果需要保留pandas的转换逻辑，也可以直接使用Beam自带的apache_beam.io.gcp.gcsio.GcsIO完成文件写入，同样不需要依赖google-cloud-storage库，但性能和稳定性不如原生WriteToText组件。

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