通过数据流将pubsub数据写入gcs

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions
from apache_beam.io.gcp.pubsub import ReadFromPubSub
from apache_beam.io.gcp.gcsio import WriteToGCS

class ParseMessage(beam.DoFn):
    def process(self, element):
        # 假设消息是以JSON格式发送的
        import json
        record = json.loads(element.decode('utf-8'))
        yield record

def run(project_id, pubsub_topic, output_path, service_account_keyfile):
    options = PipelineOptions()
    gcp_options = options.view_as(GoogleCloudOptions)
    gcp_options.project = project_id
    gcp_options.region = 'us-central1'  # 或者你选择的区域
    gcp_options.job_name = 'pubsub-to-gcs'
    gcp_options.service_account_email = service_account_keyfile
    gcp_options.use_public_ips = False

    with beam.Pipeline(options=options) as p:
        (p
         | 'Read from PubSub' >> ReadFromPubSub(topic=pubsub_topic)
         | 'Parse JSON' >> beam.ParDo(ParseMessage())
         | 'Write to GCS' >> WriteToGCS(output_path, file_naming=WriteToGCS.FileNaming.GENERATE_FILE_NAME))

if __name__ == '__main__':
    project_id = 'your-project-id'
    pubsub_topic = 'projects/your-project-id/topics/your-topic-name'
    output_path = 'gs://your-bucket-name/output/'
    service_account_keyfile = 'path/to/your/service-account-key.json'

    run(project_id, pubsub_topic, output_path, service_account_keyfile)