src/dirbs/importer/abstract_importer.py
def _upload_pipeline(self):
        """Method to handle extracting, splitting, preprocessing, pre-validating and uploading input file."""
        self._logger.info('Extracting, splitting, preprocessing, prevalidating and uploading contents from file...')
        file_to_split = self._file_to_split()

        # State variables for parallel execution
        pipeline_state = defaultdict(int)

        # The idea here is that we have 2 executors, one for prevalidation and one for upload. We do this so
        # that we can have different numbers of workers for both and so that they can happen simultaneously.
        # A job is only ever submitted for upload once it has passed pre-validation but that should happen as soon
        # as a batch has been pre-validated
        futures_to_type = {}
        with futures.ProcessPoolExecutor(max_workers=self._max_local_cpus) as prevalidator, \
                futures.ProcessPoolExecutor(max_workers=self._max_db_connections) as uploader:
            self._logger.info('Simultaneously splitting, pre-validating and uploading '
                              '({0} pre-validation workers, {1} upload workers)'
                              .format(self._max_local_cpus, self._max_db_connections))
            for f in self._split_file(file_to_split):
                # We check for any completed jobs. We do this inside the loop with a zero timeout so that we can
                # kick off any upload jobs as soon as possible. We also do this before we kick off pre-validation
                # job so that when we exit this for loop we still have some jobs pending -- we do this so that we
                # at least the final batch has a progress message printed out
                self._process_pipeline_jobs(executor=uploader, futures_to_type=futures_to_type,
                                            state=pipeline_state, timeout=0)

                # Process new split batch
                pipeline_state['num_batches'] += 1
                processed_fn = self._preprocess_file(f)  # Pre-process file
                job = prevalidator.submit(self._prevalidate_file, processed_fn)  # Kick-off pre-validation job
                futures_to_type[job] = UploadPipelineJobType.PREVALIDATE

            # At this point, we are complete with splitting, so we just need to wait until all futures are done
            while futures_to_type:
                self._process_pipeline_jobs(executor=uploader, futures_to_type=futures_to_type,
                                            state=pipeline_state, num_batches_finalized=True)

            # Calculate number of uploaded rows
            with self._conn as conn, conn.cursor() as cursor:
                cursor.execute(sql.SQL("""SELECT COUNT(*) FROM {0}""").format(self._staging_tbl_identifier))
                self._data_length = cursor.fetchone()[0]

            self._logger.info('Successully pre-validated and uploaded {0:d} rows to the staging table'
                              .format(self._data_length))
Method to handle extracting, splitting, preprocessing, pre-validating and uploading input file.
Method to handle extracting, splitting, preprocessing, pre-validating and uploading input file.
Method to handle extracting splitting preprocessing pre - validating and uploading input file .
def _upload_pipeline(self):
        self._logger.info('Extracting, splitting, preprocessing, prevalidating and uploading contents from file...')
        file_to_split = self._file_to_split()
        pipeline_state = defaultdict(int)
        futures_to_type = {}
        with futures.ProcessPoolExecutor(max_workers=self._max_local_cpus) as prevalidator, \
                futures.ProcessPoolExecutor(max_workers=self._max_db_connections) as uploader:
            self._logger.info('Simultaneously splitting, pre-validating and uploading '
                              '({0} pre-validation workers, {1} upload workers)'
                              .format(self._max_local_cpus, self._max_db_connections))
            for f in self._split_file(file_to_split):
                self._process_pipeline_jobs(executor=uploader, futures_to_type=futures_to_type,
                                            state=pipeline_state, timeout=0)
                pipeline_state['num_batches'] += 1
                processed_fn = self._preprocess_file(f)  
                job = prevalidator.submit(self._prevalidate_file, processed_fn)  
                futures_to_type[job] = UploadPipelineJobType.PREVALIDATE
            while futures_to_type:
                self._process_pipeline_jobs(executor=uploader, futures_to_type=futures_to_type,
                                            state=pipeline_state, num_batches_finalized=True)
            with self._conn as conn, conn.cursor() as cursor:
                cursor.execute(sql.SQL("""SELECT COUNT(*) FROM {0}""").format(self._staging_tbl_identifier))
                self._data_length = cursor.fetchone()[0]
            self._logger.info('Successully pre-validated and uploaded {0:d} rows to the staging table'
                              .format(self._data_length))
def _upload_pipeline ( self ) : self . _logger . info ( 'Extracting, splitting, preprocessing, prevalidating and uploading contents from file...' ) file_to_split = self . _file_to_split ( ) pipeline_state = defaultdict ( int ) futures_to_type = { } with futures . ProcessPoolExecutor ( max_workers = self . _max_local_cpus ) as prevalidator , futures . ProcessPoolExecutor ( max_workers = self . _max_db_connections ) as uploader : self . _logger . info ( 'Simultaneously splitting, pre-validating and uploading ' '({0} pre-validation workers, {1} upload workers)' . format ( self . _max_local_cpus , self . _max_db_connections ) ) for f in self . _split_file ( file_to_split ) : self . _process_pipeline_jobs ( executor = uploader , futures_to_type = futures_to_type , state = pipeline_state , timeout = 0 ) pipeline_state [ 'num_batches' ] += 1 processed_fn = self . _preprocess_file ( f ) job = prevalidator . submit ( self . _prevalidate_file , processed_fn ) futures_to_type [ job ] = UploadPipelineJobType . PREVALIDATE while futures_to_type : self . _process_pipeline_jobs ( executor = uploader , futures_to_type = futures_to_type , state = pipeline_state , num_batches_finalized = True ) with self . _conn as conn , conn . cursor ( ) as cursor : cursor . execute ( sql . SQL ( """SELECT COUNT(*) FROM {0}""" ) . format ( self . _staging_tbl_identifier ) ) self . _data_length = cursor . fetchone ( ) [ 0 ] self . _logger . info ( 'Successully pre-validated and uploaded {0:d} rows to the staging table' . format ( self . _data_length ) )
Method to handle extracting, splitting, preprocessing, pre-validating and uploading input file.
Method to handle extracting splitting preprocessing pre - validating and uploading input file .
"""Method to handle extracting, splitting, preprocessing, pre-validating and uploading input file.""" # State variables for parallel execution # The idea here is that we have 2 executors, one for prevalidation and one for upload. We do this so # that we can have different numbers of workers for both and so that they can happen simultaneously. # A job is only ever submitted for upload once it has passed pre-validation but that should happen as soon # as a batch has been pre-validated # We check for any completed jobs. We do this inside the loop with a zero timeout so that we can # kick off any upload jobs as soon as possible. We also do this before we kick off pre-validation # job so that when we exit this for loop we still have some jobs pending -- we do this so that we # at least the final batch has a progress message printed out # Process new split batch # Pre-process file # Kick-off pre-validation job # At this point, we are complete with splitting, so we just need to wait until all futures are done # Calculate number of uploaded rows
self