def parse_args(self):
    parser = argparse.ArgumentParser()
    parser.add_argument('--query', '-q', type=int, action='append', required=True, choices=[0, 1, 2], help='Query to run')
    parser.add_argument('--subscription_name', type=str, help='Pub/Sub subscription to read from')
    parser.add_argument('--topic_name', type=str, help='Pub/Sub topic to read from')
    parser.add_argument('--loglevel', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], default='INFO', help='Set logging level to debug')
    parser.add_argument('--input', type=str, required=True, help='Path to the data file containing nexmark events.')
    (self.args, self.pipeline_args) = parser.parse_known_args()
    logging.basicConfig(level=getattr(logging, self.args.loglevel, None), format='(%(threadName)-10s) %(message)s')
    self.pipeline_options = PipelineOptions(self.pipeline_args)
    logging.debug('args, pipeline_args: %s, %s', self.args, self.pipeline_args)
    self.project = self.pipeline_options.view_as(GoogleCloudOptions).project
    if (self.project is None):
        parser.print_usage()
        print((sys.argv[0] + ': error: argument --project is required'))
        sys.exit(1)
    self.streaming = self.pipeline_options.view_as(StandardOptions).streaming
    if (self.streaming is None):
        parser.print_usage()
        print((sys.argv[0] + ': error: argument --streaming is required'))
        sys.exit(1)
    self.wait_until_finish_duration = self.pipeline_options.view_as(TestOptions).wait_until_finish_duration
    if (self.wait_until_finish_duration is None):
        parser.print_usage()
        print((sys.argv[0] + ': error: argument --wait_until_finish_duration is required'))
        sys.exit(1)
    self.pipeline_options.view_as(SetupOptions).save_main_session = True
def parse_args(self):
    parser = argparse.ArgumentParser()
    parser.add_argument('--query', '-q', type=int, action='append', required=True, choices=[0, 1, 2], help='Query to run')
    parser.add_argument('--subscription_name', type=str, help='Pub/Sub subscription to read from')
    parser.add_argument('--topic_name', type=str, help='Pub/Sub topic to read from')
    parser.add_argument('--loglevel', choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'], default='INFO', help='Set logging level to debug')
    parser.add_argument('--input', type=str, required=True, help='Path to the data file containing nexmark events.')
    (self.args, self.pipeline_args) = parser.parse_known_args()
    logging.basicConfig(level=getattr(logging, self.args.loglevel, None), format='(%(threadName)-10s) %(message)s')
    self.pipeline_options = PipelineOptions(self.pipeline_args)
    logging.debug('args, pipeline_args: %s, %s', self.args, self.pipeline_args)
    self.project = self.pipeline_options.view_as(GoogleCloudOptions).project
    if (self.project is None):
        parser.print_usage()
        print((sys.argv[0] + ': error: argument --project is required'))
        sys.exit(1)
    self.streaming = self.pipeline_options.view_as(StandardOptions).streaming
    if (self.streaming is None):
        parser.print_usage()
        print((sys.argv[0] + ': error: argument --streaming is required'))
        sys.exit(1)
    self.wait_until_finish_duration = self.pipeline_options.view_as(TestOptions).wait_until_finish_duration
    if (self.wait_until_finish_duration is None):
        parser.print_usage()
        print((sys.argv[0] + ': error: argument --wait_until_finish_duration is required'))
        sys.exit(1)
    self.pipeline_options.view_as(SetupOptions).save_main_session = True