Ошибка RuntimeValueProvider при загрузке шаблона потока данных - PullRequest
Новая
       0

Ошибка RuntimeValueProvider при загрузке шаблона потока данных

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

Всякий раз, когда я пытаюсь выполнить свой шаблон конвейера с помощью add_value_provider_argument согласно инструкциям здесь: https://cloud.google.com/dataflow/docs/templates/creating-templates, конвейер пытается выполнить, вызывая ошибку, вместо загрузки в корзину GCS.

Здеськод, который я развертываю: 

#deploy

python -m main \
--runner DataflowRunner \
--project $PROJECT \
--staging_location gs://$DATAFLOW_BUCKET/staging \
--temp_location gs://$DATAFLOW_BUCKET/temp \
--output gs://$DATAFLOW_BUCKET/output \
--template_location gs://$DATAFLOW_BUCKET/templates/$TEMPLATE_NAME

#pipeline.py

class MyOptions(PipelineOptions):

  @classmethod
  def _add_argparse_args(cls, parser):      
    parser.add_value_provider_argument( #add_argument
                        '--date',
                        required = False,
                        default='2018-09-28',
                        help='Date to process, e.g. 2018-09-28'                        
                        )    


RUNNER_TYPE = 'DataflowRunner'

version = datetime.datetime.now().strftime('%Y%m%d-%H%M%S')

pipeline_options = PipelineOptions()
custom_options = pipeline_options.view_as(MyOptions)

options = PipelineOptions()

google_cloud_options = pipeline_options.view_as(GoogleCloudOptions)
google_cloud_options.project = PROJECT_ID
google_cloud_options.job_name = 'test-v{}'.format(version)
google_cloud_options.staging_location = 'gs://{}/staging'.format(STAGING_BUCKET)
google_cloud_options.temp_location = 'gs://{}/temp'.format(STAGING_BUCKET)
pipeline_options.view_as(StandardOptions).runner = RUNNER_TYPE
pipeline_options.view_as(StandardOptions).streaming = False

#installing packages used in process
setup_options = pipeline_options.view_as(SetupOptions)
setup_options.setup_file = './setup.py'
setup_options.save_main_session = False       

def run(argv=None):

    with beam.Pipeline(options=pipeline_options) as p: 

        read_file = 'gs://{}/{}-*'\
                            .format(DATA_BUCKET,custom_options.date)


        data = (p | 'Read' >> ReadFromText(read_file,coder=JsonCoder())
                  | 'ParseData' >> beam.ParDo(parse_data)
                  | 'FragmentData' >> beam.ParDo(fragment)
                  | 'CleanHeader' >> beam.ParDo(clean_header)
                  | 'RemoveMalformedRows' >> beam.ParDo(remove_malformed_rows)
                  | 'ZipData' >> beam.ParDo(zip_data)
                  | 'FilterFields' >> beam.ParDo(filter_fields)   
        )

        bigquery_write_fn = BigQueryWriteFn(table_id=TABLE_ID,dataset_id=DATASET_ID,
                                        project_id=PROJECT_ID,batch_size=100,schema=SCHEMA,
                                        create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
                                        write_disposition=BigQueryDisposition.WRITE_APPEND, #WRITE_TRUNCATE, WRITE_APPEND
                                        client=None)        

        data | 'WriteToBigQuery' >> beam.ParDo(bigquery_write_fn)           

if __name__ == '__main__':
    run()

Ошибка: 

IOError: No files found based on the file pattern gs://<BUCKET>/RuntimeValueProvider(option: date, type: str, default_value: '2018-09-28')-*

Странно, когда я использую parser.add_argument вместо parser.add_value_provider_argument, шаблон загружается в GCS, но я могуне меняйте параметр по умолчанию.

Почему конвейер выполняется вместо загрузки при изменении аргумента синтаксического анализатора с add_argument на add_value_provider_argument.

1 Ответ

0 голосов
/ 30 марта 2019

Недавно я столкнулся с той же проблемой, и проблема в том, что объекты ValueProvider недоступны во время построения конвейера.Это означает, что в Python вы не можете задавать имена файлов или создавать динамические имена файлов на основе RuntimeValueProviders.

Вместо этого вам нужно добавить еще один аргумент, который заменяет переменную read_file и передается непосредственно в метод ReadFromText.

class MyOptions(PipelineOptions):

  @classmethod
  def _add_argparse_args(cls, parser):      
    parser.add_value_provider_argument( #add_argument
                        '--input',
                        required = False,
                        default='',
                        help='Full path to input file'                        
                        )    


RUNNER_TYPE = 'DataflowRunner'

version = datetime.datetime.now().strftime('%Y%m%d-%H%M%S')

pipeline_options = PipelineOptions()
custom_options = pipeline_options.view_as(MyOptions)

options = PipelineOptions()

google_cloud_options = pipeline_options.view_as(GoogleCloudOptions)
google_cloud_options.project = PROJECT_ID
google_cloud_options.job_name = 'test-v{}'.format(version)
google_cloud_options.staging_location = 'gs://{}/staging'.format(STAGING_BUCKET)
google_cloud_options.temp_location = 'gs://{}/temp'.format(STAGING_BUCKET)
pipeline_options.view_as(StandardOptions).runner = RUNNER_TYPE
pipeline_options.view_as(StandardOptions).streaming = False

#installing packages used in process
setup_options = pipeline_options.view_as(SetupOptions)
setup_options.setup_file = './setup.py'
setup_options.save_main_session = False       

def run():

    with beam.Pipeline(options=pipeline_options) as p: 

        data = (p | 'Read' >> ReadFromText(custom_options.input,coder=JsonCoder())
                  | 'ParseData' >> beam.ParDo(parse_data)
                  | 'FragmentData' >> beam.ParDo(fragment)
                  | 'CleanHeader' >> beam.ParDo(clean_header)
                  | 'RemoveMalformedRows' >> beam.ParDo(remove_malformed_rows)
                  | 'ZipData' >> beam.ParDo(zip_data)
                  | 'FilterFields' >> beam.ParDo(filter_fields)   
        )

        bigquery_write_fn = BigQueryWriteFn(table_id=TABLE_ID,dataset_id=DATASET_ID,
                                        project_id=PROJECT_ID,batch_size=100,schema=SCHEMA,
                                        create_disposition=BigQueryDisposition.CREATE_IF_NEEDED,
                                        write_disposition=BigQueryDisposition.WRITE_APPEND, #WRITE_TRUNCATE, WRITE_APPEND
                                        client=None)        

        data | 'WriteToBigQuery' >> beam.ParDo(bigquery_write_fn)           

if __name__ == '__main__':
    run()
...