Source code for apache_beam.io.gcp.pubsub_it_pipeline

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
"""
Test pipeline for use by pubsub_integration_test.
"""

from __future__ import absolute_import

import argparse

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions
from apache_beam.options.pipeline_options import StandardOptions


[docs]def run_pipeline(argv, with_attributes, id_label, timestamp_attribute): """Build and run the pipeline.""" parser = argparse.ArgumentParser() parser.add_argument( '--output_topic', required=True, help=('Output PubSub topic of the form ' '"projects/<PROJECT>/topic/<TOPIC>".')) parser.add_argument( '--input_subscription', required=True, help=('Input PubSub subscription of the form ' '"projects/<PROJECT>/subscriptions/<SUBSCRIPTION>."')) known_args, pipeline_args = parser.parse_known_args(argv) # We use the save_main_session option because one or more DoFn's in this # workflow rely on global context (e.g., a module imported at module level). pipeline_options = PipelineOptions(pipeline_args) pipeline_options.view_as(SetupOptions).save_main_session = True pipeline_options.view_as(StandardOptions).streaming = True p = beam.Pipeline(options=pipeline_options) runner_name = type(p.runner).__name__ # Read from PubSub into a PCollection. if runner_name == 'TestDirectRunner': messages = p | beam.io.ReadFromPubSub( subscription=known_args.input_subscription, with_attributes=with_attributes, timestamp_attribute=timestamp_attribute) else: messages = p | beam.io.ReadFromPubSub( subscription=known_args.input_subscription, id_label=id_label, with_attributes=with_attributes, timestamp_attribute=timestamp_attribute) def add_attribute(msg, timestamp=beam.DoFn.TimestampParam): msg.data += b'-seen' msg.attributes['processed'] = 'IT' if timestamp_attribute in msg.attributes: msg.attributes[timestamp_attribute + '_out'] = timestamp.to_rfc3339() return msg def modify_data(data): return data + b'-seen' if with_attributes: output = messages | 'add_attribute' >> beam.Map(add_attribute) else: output = messages | 'modify_data' >> beam.Map(modify_data) # Write to PubSub. if runner_name == 'TestDirectRunner': _ = output | beam.io.WriteToPubSub(known_args.output_topic, with_attributes=with_attributes) else: _ = output | beam.io.WriteToPubSub(known_args.output_topic, id_label=id_label, with_attributes=with_attributes, timestamp_attribute=timestamp_attribute) result = p.run() result.wait_until_finish()