def run(args, pipeline_args=None):
    'Executes Pipeline.\n\n    :param args:\n    :param pipeline_args:\n    :return:\n    '
    'Build and run the pipeline.'
    pipeline_options = PipelineOptions(pipeline_args, streaming=True, save_main_session=True)
    pipeline_options.view_as(StandardOptions).runner = args.runner
    google_cloud_options = pipeline_options.view_as(GoogleCloudOptions)
    google_cloud_options.project = PROJECT_ID
    google_cloud_options.job_name = 'pubsub-api-bigquery'
    google_cloud_options.staging_location = args.staging_location
    google_cloud_options.temp_location = args.temp_location
    google_cloud_options.region = args.region
    p = beam.Pipeline(options=pipeline_options)
    lines = (p | ('read in tweets' >> beam.io.ReadFromPubSub(topic=args.input_topic, with_attributes=False, id_label='tweet_id')))
    output_tweets = (((lines | ('assign window key' >> beam.WindowInto(window.FixedWindows(args.window_size)))) | ('batch into n batches' >> BatchElements(min_batch_size=args.min_batch_size, max_batch_size=args.max_batch_size))) | ('predict sentiment' >> beam.FlatMap((lambda messages: prediction_helper(messages)))))
    bq_schema_json = {'fields': [{'name': 'id', 'type': 'STRING'}, {'name': 'text', 'type': 'STRING'}, {'name': 'user_id', 'type': 'STRING'}, {'name': 'sentiment', 'type': 'FLOAT'}, {'name': 'posted_at', 'type': 'TIMESTAMP'}, {'name': 'favorite_count', 'type': 'INTEGER'}, {'name': 'retweet_count', 'type': 'INTEGER'}, {'name': 'media', 'type': 'STRING'}]}
    bq_schema = parse_table_schema_from_json(json.dumps(bq_schema_json))
    (output_tweets | ('store twitter posts' >> beam.io.WriteToBigQuery(table=args.bigquery_table, dataset=args.bigquery_dataset, schema=bq_schema, write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND, create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, project=PROJECT_ID)))
    result = p.run()
    result.wait_until_finish()
Executes Pipeline.

:param args:
:param pipeline_args:
:return:
def run(args, pipeline_args=None):
    'Executes Pipeline.\n\n    :param args:\n    :param pipeline_args:\n    :return:\n    '
    'Build and run the pipeline.'
    pipeline_options = PipelineOptions(pipeline_args, streaming=True, save_main_session=True)
    pipeline_options.view_as(StandardOptions).runner = args.runner
    google_cloud_options = pipeline_options.view_as(GoogleCloudOptions)
    google_cloud_options.project = PROJECT_ID
    google_cloud_options.job_name = 'pubsub-api-bigquery'
    google_cloud_options.staging_location = args.staging_location
    google_cloud_options.temp_location = args.temp_location
    google_cloud_options.region = args.region
    p = beam.Pipeline(options=pipeline_options)
    lines = (p | ('read in tweets' >> beam.io.ReadFromPubSub(topic=args.input_topic, with_attributes=False, id_label='tweet_id')))
    output_tweets = (((lines | ('assign window key' >> beam.WindowInto(window.FixedWindows(args.window_size)))) | ('batch into n batches' >> BatchElements(min_batch_size=args.min_batch_size, max_batch_size=args.max_batch_size))) | ('predict sentiment' >> beam.FlatMap((lambda messages: prediction_helper(messages)))))
    bq_schema_json = {'fields': [{'name': 'id', 'type': 'STRING'}, {'name': 'text', 'type': 'STRING'}, {'name': 'user_id', 'type': 'STRING'}, {'name': 'sentiment', 'type': 'FLOAT'}, {'name': 'posted_at', 'type': 'TIMESTAMP'}, {'name': 'favorite_count', 'type': 'INTEGER'}, {'name': 'retweet_count', 'type': 'INTEGER'}, {'name': 'media', 'type': 'STRING'}]}
    bq_schema = parse_table_schema_from_json(json.dumps(bq_schema_json))
    (output_tweets | ('store twitter posts' >> beam.io.WriteToBigQuery(table=args.bigquery_table, dataset=args.bigquery_dataset, schema=bq_schema, write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND, create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED, project=PROJECT_ID)))
    result = p.run()
    result.wait_until_finish()