src/dirbs/importer/base_delta_importer.py
def _compute_full_list_to_delta(self, executor):
        """Generator to compute the delta from a full list, yielding results as each type of delta is complete."""
        futures_to_type = {}
        num_physical_shards = partition_utils.num_physical_imei_shards(self._conn) if self._supports_imei_shards else 1
        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)
                import_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)
                delta_tbl_part_name = partition_utils.imei_shard_name(base_name=self._delta_tbl_name,
                                                                      virt_imei_range_start=virt_imei_range_start,
                                                                      virt_imei_range_end=virt_imei_range_end)

                f = executor.submit(self._calc_delta_removes_single_shard,
                                    staging_tbl_part_name=staging_tbl_part_name,
                                    import_tbl_part_name=import_tbl_part_name,
                                    delta_tbl_part_name=delta_tbl_part_name,
                                    import_filter='end_date IS NULL')
                futures_to_type[f] = 'removes'
                f = executor.submit(self._calc_delta_adds_single_shard,
                                    staging_tbl_part_name=staging_tbl_part_name,
                                    import_tbl_part_name=import_tbl_part_name,
                                    delta_tbl_part_name=delta_tbl_part_name,
                                    import_filter='end_date IS NULL')
                futures_to_type[f] = 'adds'
                f = executor.submit(self._calc_delta_updates_single_shard,
                                    staging_tbl_part_name=staging_tbl_part_name,
                                    import_tbl_part_name=import_tbl_part_name,
                                    delta_tbl_part_name=delta_tbl_part_name,
                                    import_filter='end_date IS NULL')
                futures_to_type[f] = 'updates'
        else:
            f = executor.submit(self._calc_delta_removes_single_shard,
                                staging_tbl_part_name=self._staging_tbl_name,
                                import_tbl_part_name=self._import_relation_name,
                                delta_tbl_part_name=self._delta_tbl_name)
            futures_to_type[f] = 'removes'
            f = executor.submit(self._calc_delta_adds_single_shard,
                                staging_tbl_part_name=self._staging_tbl_name,
                                import_tbl_part_name=self._import_relation_name,
                                delta_tbl_part_name=self._delta_tbl_name)
            futures_to_type[f] = 'adds'
            f = executor.submit(self._calc_delta_updates_single_shard,
                                staging_tbl_part_name=self._staging_tbl_name,
                                import_tbl_part_name=self._import_relation_name,
                                delta_tbl_part_name=self._delta_tbl_name)
            futures_to_type[f] = 'updates'

        job_completion_state = dict(adds=0, removes=0, updates=0)
        rows_changed_state = dict(adds=0, removes=0, updates=0)
        for f in futures.as_completed(futures_to_type):
            rows_changed = f.result()
            change_type = futures_to_type[f]
            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 to compute the delta from a full list, yielding results as each type of delta is complete.
Generator to compute the delta from a full list, yielding results as each type of delta is complete.
Generator to compute the delta from a full list yielding results as each type of delta is complete .
def _compute_full_list_to_delta(self, executor):
        futures_to_type = {}
        num_physical_shards = partition_utils.num_physical_imei_shards(self._conn) if self._supports_imei_shards else 1
        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)
                import_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)
                delta_tbl_part_name = partition_utils.imei_shard_name(base_name=self._delta_tbl_name,
                                                                      virt_imei_range_start=virt_imei_range_start,
                                                                      virt_imei_range_end=virt_imei_range_end)
                f = executor.submit(self._calc_delta_removes_single_shard,
                                    staging_tbl_part_name=staging_tbl_part_name,
                                    import_tbl_part_name=import_tbl_part_name,
                                    delta_tbl_part_name=delta_tbl_part_name,
                                    import_filter='end_date IS NULL')
                futures_to_type[f] = 'removes'
                f = executor.submit(self._calc_delta_adds_single_shard,
                                    staging_tbl_part_name=staging_tbl_part_name,
                                    import_tbl_part_name=import_tbl_part_name,
                                    delta_tbl_part_name=delta_tbl_part_name,
                                    import_filter='end_date IS NULL')
                futures_to_type[f] = 'adds'
                f = executor.submit(self._calc_delta_updates_single_shard,
                                    staging_tbl_part_name=staging_tbl_part_name,
                                    import_tbl_part_name=import_tbl_part_name,
                                    delta_tbl_part_name=delta_tbl_part_name,
                                    import_filter='end_date IS NULL')
                futures_to_type[f] = 'updates'
        else:
            f = executor.submit(self._calc_delta_removes_single_shard,
                                staging_tbl_part_name=self._staging_tbl_name,
                                import_tbl_part_name=self._import_relation_name,
                                delta_tbl_part_name=self._delta_tbl_name)
            futures_to_type[f] = 'removes'
            f = executor.submit(self._calc_delta_adds_single_shard,
                                staging_tbl_part_name=self._staging_tbl_name,
                                import_tbl_part_name=self._import_relation_name,
                                delta_tbl_part_name=self._delta_tbl_name)
            futures_to_type[f] = 'adds'
            f = executor.submit(self._calc_delta_updates_single_shard,
                                staging_tbl_part_name=self._staging_tbl_name,
                                import_tbl_part_name=self._import_relation_name,
                                delta_tbl_part_name=self._delta_tbl_name)
            futures_to_type[f] = 'updates'
        job_completion_state = dict(adds=0, removes=0, updates=0)
        rows_changed_state = dict(adds=0, removes=0, updates=0)
        for f in futures.as_completed(futures_to_type):
            rows_changed = f.result()
            change_type = futures_to_type[f]
            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 _compute_full_list_to_delta ( self , executor ) : futures_to_type = { } num_physical_shards = partition_utils . num_physical_imei_shards ( self . _conn ) if self . _supports_imei_shards else 1 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 ) import_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 ) delta_tbl_part_name = partition_utils . imei_shard_name ( base_name = self . _delta_tbl_name , virt_imei_range_start = virt_imei_range_start , virt_imei_range_end = virt_imei_range_end ) f = executor . submit ( self . _calc_delta_removes_single_shard , staging_tbl_part_name = staging_tbl_part_name , import_tbl_part_name = import_tbl_part_name , delta_tbl_part_name = delta_tbl_part_name , import_filter = 'end_date IS NULL' ) futures_to_type [ f ] = 'removes' f = executor . submit ( self . _calc_delta_adds_single_shard , staging_tbl_part_name = staging_tbl_part_name , import_tbl_part_name = import_tbl_part_name , delta_tbl_part_name = delta_tbl_part_name , import_filter = 'end_date IS NULL' ) futures_to_type [ f ] = 'adds' f = executor . submit ( self . _calc_delta_updates_single_shard , staging_tbl_part_name = staging_tbl_part_name , import_tbl_part_name = import_tbl_part_name , delta_tbl_part_name = delta_tbl_part_name , import_filter = 'end_date IS NULL' ) futures_to_type [ f ] = 'updates' else : f = executor . submit ( self . _calc_delta_removes_single_shard , staging_tbl_part_name = self . _staging_tbl_name , import_tbl_part_name = self . _import_relation_name , delta_tbl_part_name = self . _delta_tbl_name ) futures_to_type [ f ] = 'removes' f = executor . submit ( self . _calc_delta_adds_single_shard , staging_tbl_part_name = self . _staging_tbl_name , import_tbl_part_name = self . _import_relation_name , delta_tbl_part_name = self . _delta_tbl_name ) futures_to_type [ f ] = 'adds' f = executor . submit ( self . _calc_delta_updates_single_shard , staging_tbl_part_name = self . _staging_tbl_name , import_tbl_part_name = self . _import_relation_name , delta_tbl_part_name = self . _delta_tbl_name ) futures_to_type [ f ] = 'updates' job_completion_state = dict ( adds = 0 , removes = 0 , updates = 0 ) rows_changed_state = dict ( adds = 0 , removes = 0 , updates = 0 ) for f in futures . as_completed ( futures_to_type ) : rows_changed = f . result ( ) change_type = futures_to_type [ f ] 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 to compute the delta from a full list, yielding results as each type of delta is complete.
Generator to compute the delta from a full list yielding results as each type of delta is complete .
"""Generator to compute the delta from a full list, yielding results as each type of delta is complete."""
self
executor