您似乎希望根据ManagerName列将数据写入GCP。您可以使用Apache Beam中的Google Cloud Storage（GCS）API实现此操作。请尝试以下操作：

from apache_beam.io.gcp.internal.clients import bigquery
from apache_beam.io.gcp.internal.clients import storage
from apache_beam.io.gcp.internal.clients.storage import StorageObjects
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam import DoFn, ParDo, GroupByKey, Pipeline

import json

class writeEachGroupToGCP(DoFn):
  def __init__(self, gcs_path):
    self.gcs_path = gcs_path

  def process(self, data):
    (manager, rows) = data
    with storage.Client().bucket(self.gcs_path).blob("Manager={}/data.json".format(manager)).open('w') as f:
      for row in rows:
        f.write(json.dumps(row) + '\n')
    return data

options = PipelineOptions()
gcp_project = options.view_as(GoogleCloudOptions).project
gcp_staging_location = options.view_as(GoogleCloudOptions).staging_location
gcp_temp_location = options.view_as(GoogleCloudOptions).temp_location
gcp_input_path = "gs://<your-bucket>/<your-input-path>"
gcp_output_path = "gs://<your-bucket>/<your-output-path>"

with Pipeline(options=options) as p1:
  (
    p1
    | "Read Data" >> beam.io.ReadFromText(gcp_input_path)
    | "Parse CSV" >> beam.Map(lambda x: tuple(x.split(",")))
    | "Group By Manager" >> beam.GroupByKey()
    | "Write To GCS" >> beam.ParDo(writeEachGroupToGCP(gcp_output_path))
  )

第四章