Именование таблицы BigQuery из параметров времени выполнения шаблона, Python, Apache Beam, Dataflow - PullRequest
Новая
       34

Именование таблицы BigQuery из параметров времени выполнения шаблона, Python, Apache Beam, Dataflow

0 голосов
/ 11 сентября 2018

Я работаю над проектом в потоке данных Python Apache Beam, и мне нужно назвать таблицы больших запросов из параметров времени выполнения, предоставленных при запуске шаблона потока данных .

Пока мне не повезло, он либо дает мне определение параметра времени выполнения, либо пустой строки.

Так что мне нужно, чтобы это как-то работало: 

class CustomPipelineOptions(PipelineOptions):

    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument(
            '--path',
            type=str,
            help='csv storage path')
        parser.add_value_provider_argument(
            '--table_name',
            type=str,
            help='Table Id')
def run()
    def rewrite_values(element):
        """ Rewrite default env values"""
        try:
            logging.info("File Path with str(): {}".format(str(custom_options.path)))
            logging.info("----------------------------")
            logging.info("element: {}".format(element))
            project_id = str(cloud_options.project)
            file_path = custom_options.path.get()
            table_name = custom_options.table_name.get()

            logging.info("project: {}".format(project_id))
            logging.info("File path: {}".format(file_path))
            logging.info("language: {}".format(table_name))
            logging.info("----------------------------")
        except Exception as e:
            logging.info("Error format----------------------------")
            raise KeyError(e)

        return file_path

    pipeline_options = PipelineOptions()
    cloud_options = pipeline_options.view_as(GoogleCloudOptions)
    custom_options = pipeline_options.view_as(CustomPipelineOptions)
    pipeline_options.view_as(SetupOptions).save_main_session = True

    # Beginning of the pipeline
    p = beam.Pipeline(options=pipeline_options)

    init_data = (p
                 | beam.Create(["Start"])
                 | beam.FlatMap(rewrite_values))

pipeline processing, running pipeline etc.

save_data_bigquery = (table_data | "Get all numbers" >> beam.ParDo(GetAllNumbers())
                      | 'Flat items' >> beam.FlatMap(flat_item)
                      | 'Write to BigQuery' >> beam.io.WriteToBigQuery(project=project_id,
                                                                       dataset="defined_dataset",
                                                                       table=table_name, **********
                                                                       schema="id:STRING",
                                                                       batch_size=10000)
                      )

При именовании таблицы в функции writetobigquery у меня возникают проблемы, я также пытался использовать custom_options.table_name, объявляя переменные как глобальные и т. Д. И т. Д.

Я создал собственный DoFn для записи в BigQuery, хотя это мой предпочтительный метод.

...