src/dirbs/importer/base_delta_importer.py
def _populate_historic_from_delta(self, executor, job_start_time):
        """Generator which processes all the updates to the historic table and yields when a category is complete."""
        num_physical_shards = partition_utils.num_physical_imei_shards(self._conn) if self._supports_imei_shards else 1
        futures_list = []
        if self._supports_imei_shards:
            virt_imei_shard_ranges = partition_utils.virt_imei_shard_bounds(num_physical_shards)
            for virt_imei_range_start, virt_imei_range_end in virt_imei_shard_ranges:
                staging_tbl_part_name = partition_utils.imei_shard_name(base_name=self._staging_tbl_name,
                                                                        virt_imei_range_start=virt_imei_range_start,
                                                                        virt_imei_range_end=virt_imei_range_end)
                historic_tbl_part_name = partition_utils.imei_shard_name(base_name=self._historic_tbl_name,
                                                                         virt_imei_range_start=virt_imei_range_start,
                                                                         virt_imei_range_end=virt_imei_range_end)

                futures_list.append(executor.submit(self._add_into_historic_single_shard,
                                                    staging_tbl_part_name,
                                                    historic_tbl_part_name,
                                                    job_start_time))
                futures_list.append(executor.submit(self._remove_from_historic_single_shard,
                                                    staging_tbl_part_name,
                                                    historic_tbl_part_name,
                                                    job_start_time))
                futures_list.append(executor.submit(self._update_historic_single_shard,
                                                    staging_tbl_part_name,
                                                    historic_tbl_part_name,
                                                    job_start_time))
        else:
            futures_list.append(executor.submit(self._add_into_historic_single_shard,
                                                self._staging_tbl_name,
                                                self._historic_tbl_name,
                                                job_start_time))
            futures_list.append(executor.submit(self._remove_from_historic_single_shard,
                                                self._staging_tbl_name,
                                                self._historic_tbl_name,
                                                job_start_time))
            futures_list.append(executor.submit(self._update_historic_single_shard,
                                                self._staging_tbl_name,
                                                self._historic_tbl_name,
                                                job_start_time))

        job_completion_state = dict(add=0, update=0, remove=0)
        rows_changed_state = dict(add=0, update=0, remove=0)
        for f in futures.as_completed(futures_list):
            change_type, rows_changed = f.result()
            job_completion_state[change_type] += 1
            rows_changed_state[change_type] += rows_changed
            if job_completion_state[change_type] == num_physical_shards:
                yield change_type, rows_changed_state[change_type]
Generator which processes all the updates to the historic table and yields when a category is complete.
Generator which processes all the updates to the historic table and yields when a category is complete.
Generator which processes all the updates to the historic table and yields when a category is complete .
def _populate_historic_from_delta(self, executor, job_start_time):
        num_physical_shards = partition_utils.num_physical_imei_shards(self._conn) if self._supports_imei_shards else 1
        futures_list = []
        if self._supports_imei_shards:
            virt_imei_shard_ranges = partition_utils.virt_imei_shard_bounds(num_physical_shards)
            for virt_imei_range_start, virt_imei_range_end in virt_imei_shard_ranges:
                staging_tbl_part_name = partition_utils.imei_shard_name(base_name=self._staging_tbl_name,
                                                                        virt_imei_range_start=virt_imei_range_start,
                                                                        virt_imei_range_end=virt_imei_range_end)
                historic_tbl_part_name = partition_utils.imei_shard_name(base_name=self._historic_tbl_name,
                                                                         virt_imei_range_start=virt_imei_range_start,
                                                                         virt_imei_range_end=virt_imei_range_end)
                futures_list.append(executor.submit(self._add_into_historic_single_shard,
                                                    staging_tbl_part_name,
                                                    historic_tbl_part_name,
                                                    job_start_time))
                futures_list.append(executor.submit(self._remove_from_historic_single_shard,
                                                    staging_tbl_part_name,
                                                    historic_tbl_part_name,
                                                    job_start_time))
                futures_list.append(executor.submit(self._update_historic_single_shard,
                                                    staging_tbl_part_name,
                                                    historic_tbl_part_name,
                                                    job_start_time))
        else:
            futures_list.append(executor.submit(self._add_into_historic_single_shard,
                                                self._staging_tbl_name,
                                                self._historic_tbl_name,
                                                job_start_time))
            futures_list.append(executor.submit(self._remove_from_historic_single_shard,
                                                self._staging_tbl_name,
                                                self._historic_tbl_name,
                                                job_start_time))
            futures_list.append(executor.submit(self._update_historic_single_shard,
                                                self._staging_tbl_name,
                                                self._historic_tbl_name,
                                                job_start_time))
        job_completion_state = dict(add=0, update=0, remove=0)
        rows_changed_state = dict(add=0, update=0, remove=0)
        for f in futures.as_completed(futures_list):
            change_type, rows_changed = f.result()
            job_completion_state[change_type] += 1
            rows_changed_state[change_type] += rows_changed
            if job_completion_state[change_type] == num_physical_shards:
                yield change_type, rows_changed_state[change_type]
def _populate_historic_from_delta ( self , executor , job_start_time ) : num_physical_shards = partition_utils . num_physical_imei_shards ( self . _conn ) if self . _supports_imei_shards else 1 futures_list = [ ] if self . _supports_imei_shards : virt_imei_shard_ranges = partition_utils . virt_imei_shard_bounds ( num_physical_shards ) for virt_imei_range_start , virt_imei_range_end in virt_imei_shard_ranges : staging_tbl_part_name = partition_utils . imei_shard_name ( base_name = self . _staging_tbl_name , virt_imei_range_start = virt_imei_range_start , virt_imei_range_end = virt_imei_range_end ) historic_tbl_part_name = partition_utils . imei_shard_name ( base_name = self . _historic_tbl_name , virt_imei_range_start = virt_imei_range_start , virt_imei_range_end = virt_imei_range_end ) futures_list . append ( executor . submit ( self . _add_into_historic_single_shard , staging_tbl_part_name , historic_tbl_part_name , job_start_time ) ) futures_list . append ( executor . submit ( self . _remove_from_historic_single_shard , staging_tbl_part_name , historic_tbl_part_name , job_start_time ) ) futures_list . append ( executor . submit ( self . _update_historic_single_shard , staging_tbl_part_name , historic_tbl_part_name , job_start_time ) ) else : futures_list . append ( executor . submit ( self . _add_into_historic_single_shard , self . _staging_tbl_name , self . _historic_tbl_name , job_start_time ) ) futures_list . append ( executor . submit ( self . _remove_from_historic_single_shard , self . _staging_tbl_name , self . _historic_tbl_name , job_start_time ) ) futures_list . append ( executor . submit ( self . _update_historic_single_shard , self . _staging_tbl_name , self . _historic_tbl_name , job_start_time ) ) job_completion_state = dict ( add = 0 , update = 0 , remove = 0 ) rows_changed_state = dict ( add = 0 , update = 0 , remove = 0 ) for f in futures . as_completed ( futures_list ) : change_type , rows_changed = f . result ( ) job_completion_state [ change_type ] += 1 rows_changed_state [ change_type ] += rows_changed if job_completion_state [ change_type ] == num_physical_shards : yield change_type , rows_changed_state [ change_type ]
Generator which processes all the updates to the historic table and yields when a category is complete.
Generator which processes all the updates to the historic table and yields when a category is complete .
"""Generator which processes all the updates to the historic table and yields when a category is complete."""
self
executor
job_start_time