src/dirbs/importer/operator_data_importer.py
def _update_monthly_network_triplets(self, month, year, src_partition, virt_imei_shard_start,
                                         virt_imei_shard_end):
        """Helper function to update the monthly_network_triplets tables (country and per-MNO)."""
        with create_db_connection(self._db_config) as conn, conn.cursor() as cursor:
            start_date, end_date = self._date_range_for_month_year(month, year)

            aggregated_data_temp_table = '{0}_aggregated_{1:02d}_{2:d}'.format(src_partition, month, year)
            cursor.execute(sql.SQL("""CREATE TEMPORARY TABLE {0} (LIKE monthly_network_triplets_per_mno)""")
                           .format(sql.Identifier(aggregated_data_temp_table)))

            if self._perform_msisdn_import:
                msisdn_output = sql.Identifier('msisdn_norm')
            else:
                msisdn_output = sql.SQL('NULL AS msisdn_norm')

            cursor.execute(sql.SQL("""
                INSERT INTO {0} (triplet_year, triplet_month, first_seen, last_seen, date_bitmask,
                                 triplet_hash, imei_norm, imsi, msisdn, operator_id, virt_imei_shard)
                     SELECT %s,
                            %s,
                            triplet_sq.first_seen,
                            triplet_sq.last_seen,
                            triplet_sq.date_bitmask,
                            triplet_sq.triplet_hash,
                            triplet_sq.imei_norm,
                            triplet_sq.imsi_norm,
                            triplet_sq.msisdn_norm,
                            %s,
                            calc_virt_imei_shard(triplet_sq.imei_norm)
                       FROM (SELECT hash_triplet(imei_norm, imsi_norm, msisdn_norm) AS triplet_hash,
                                    imei_norm,
                                    imsi_norm,
                                    {msisdn_output},
                                    MIN(connection_date) AS first_seen,
                                    MAX(connection_date) AS last_seen,
                                    bit_or(1 << (date_part('day', connection_date)::INT - 1)) AS date_bitmask
                               FROM (SELECT *
                                       FROM {1}
                                      WHERE connection_date >= %s
                                        AND connection_date < %s) op_data_sq
                           GROUP BY imei_norm, imsi_norm, msisdn_norm) triplet_sq""")
                           .format(sql.Identifier(aggregated_data_temp_table),
                                   sql.Identifier(src_partition),
                                   msisdn_output=msisdn_output),
                           [year, month, self._operator_id, start_date, end_date])

            # Save how many triplets actually got inserted for later on when we print stats
            self._distinct_triplet_count += cursor.rowcount

            # Need to insert into both country and per-MNO tables. Do country first
            base_partition = partition_utils.monthly_network_triplets_country_partition(month=month, year=year)
            dest_partition = partition_utils.imei_shard_name(base_name=base_partition,
                                                             virt_imei_range_start=virt_imei_shard_start,
                                                             virt_imei_range_end=virt_imei_shard_end)

            # The on conflict clause is common to both insertions
            on_conflict_sql = sql.SQL(
                """ON CONFLICT (triplet_hash)
                     DO UPDATE
                           SET first_seen = least(target.first_seen, excluded.first_seen),
                               last_seen = greatest(target.last_seen, excluded.last_seen),
                               date_bitmask = target.date_bitmask | excluded.date_bitmask
                         WHERE (target.date_bitmask | excluded.date_bitmask) != target.date_bitmask
                """  # noqa: Q441
            )

            cursor.execute(
                sql.SQL(
                    """INSERT INTO {0} AS target(triplet_year, triplet_month, first_seen, last_seen, date_bitmask,
                                                 triplet_hash, imei_norm, imsi, msisdn, virt_imei_shard)
                            SELECT triplet_year, triplet_month, first_seen, last_seen, date_bitmask, triplet_hash,
                                   imei_norm, imsi, msisdn, virt_imei_shard
                              FROM {1}
                                   {2}
                    """
                ).format(sql.Identifier(dest_partition), sql.Identifier(aggregated_data_temp_table), on_conflict_sql))

            # Now insert into the per-MNO table
            base_partition = partition_utils.monthly_network_triplets_per_mno_partition(month=month, year=year,
                                                                                        operator_id=self._operator_id)
            dest_partition = partition_utils.imei_shard_name(base_name=base_partition,
                                                             virt_imei_range_start=virt_imei_shard_start,
                                                             virt_imei_range_end=virt_imei_shard_end)
            cursor.execute(
                sql.SQL(
                    """INSERT INTO {0} AS target
                            SELECT *
                              FROM {1}
                                   {2}
                    """
                ).format(sql.Identifier(dest_partition), sql.Identifier(aggregated_data_temp_table), on_conflict_sql))

            inserted_or_updated_triplet_count = cursor.rowcount

            # Update daily_per_mno_hll_sketches table
            hll_partition_base_name = '{0}_{1:02d}_{2:d}'.format(self._staging_hll_sketches_tbl_name, month, year)
            hll_partition_name = partition_utils.imei_shard_name(base_name=hll_partition_base_name,
                                                                 virt_imei_range_start=virt_imei_shard_start,
                                                                 virt_imei_range_end=virt_imei_shard_end)
            cursor.execute(sql.SQL("""CREATE UNLOGGED TABLE {0} (LIKE {1}) INHERITS ({1})""")
                           .format(sql.Identifier(hll_partition_name),
                                   self._staging_hll_sketches_tbl_id))

            cursor.execute(sql.SQL("""
                INSERT INTO {0} (triplet_hll, imei_hll, imsi_hll, msisdn_hll, imei_imsis_hll, imei_msisdns_hll,
                                 imsi_msisdns_hll, creation_date, operator_id, data_date)
                     SELECT coalesce(hll_add_agg(hll_hash_text(hash_triplet(imei_norm, imsi, msisdn)::TEXT))
                                     filter(WHERE imei_norm IS NOT NULL
                                              AND imsi_norm IS NOT NULL
                                              AND msisdn_norm IS NOT NULL), hll_empty()) AS triplet_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm)), hll_empty()) AS imei_hll,
                            coalesce(hll_add_agg(hll_hash_text(imsi_norm)), hll_empty()) as imsi_hll,
                            coalesce(hll_add_agg(hll_hash_text(msisdn_norm)), hll_empty()) AS msisdn_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm||'$'||imsi_norm)), hll_empty())
                                AS imei_imsis_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm||'$'||msisdn_norm)), hll_empty())
                                AS imei_msisdns_hll,
                            coalesce(hll_add_agg(hll_hash_text(imsi_norm||'$'||msisdn_norm)), hll_empty())
                                AS imsi_msisdns_hll,
                            CURRENT_DATE AS creation_date,
                            %s AS operator_id,
                            make_date(%s, %s, date_part('day', connection_date)::INT) AS data_date
                       FROM {1}
                      WHERE connection_date >= %s
                        AND connection_date < %s
                   GROUP BY connection_date;
                """).format(sql.Identifier(hll_partition_name), # noqa Q441, Q447
                            sql.Identifier(src_partition)),
                           [self._operator_id, year, month, start_date, end_date])

        return inserted_or_updated_triplet_count
Helper function to update the monthly_network_triplets tables (country and per-MNO).
Helper function to update the monthly_network_triplets tables (country and per-MNO).
Helper function to update the monthly_network_triplets tables ( country and per - MNO ) .
def _update_monthly_network_triplets(self, month, year, src_partition, virt_imei_shard_start,
                                         virt_imei_shard_end):
        with create_db_connection(self._db_config) as conn, conn.cursor() as cursor:
            start_date, end_date = self._date_range_for_month_year(month, year)
            aggregated_data_temp_table = '{0}_aggregated_{1:02d}_{2:d}'.format(src_partition, month, year)
            cursor.execute(sql.SQL("""CREATE TEMPORARY TABLE {0} (LIKE monthly_network_triplets_per_mno)""")
                           .format(sql.Identifier(aggregated_data_temp_table)))
            if self._perform_msisdn_import:
                msisdn_output = sql.Identifier('msisdn_norm')
            else:
                msisdn_output = sql.SQL('NULL AS msisdn_norm')
            cursor.execute(sql.SQL("""
                INSERT INTO {0} (triplet_year, triplet_month, first_seen, last_seen, date_bitmask,
                                 triplet_hash, imei_norm, imsi, msisdn, operator_id, virt_imei_shard)
                     SELECT %s,
                            %s,
                            triplet_sq.first_seen,
                            triplet_sq.last_seen,
                            triplet_sq.date_bitmask,
                            triplet_sq.triplet_hash,
                            triplet_sq.imei_norm,
                            triplet_sq.imsi_norm,
                            triplet_sq.msisdn_norm,
                            %s,
                            calc_virt_imei_shard(triplet_sq.imei_norm)
                       FROM (SELECT hash_triplet(imei_norm, imsi_norm, msisdn_norm) AS triplet_hash,
                                    imei_norm,
                                    imsi_norm,
                                    {msisdn_output},
                                    MIN(connection_date) AS first_seen,
                                    MAX(connection_date) AS last_seen,
                                    bit_or(1 << (date_part('day', connection_date)::INT - 1)) AS date_bitmask
                               FROM (SELECT *
                                       FROM {1}
                                      WHERE connection_date >= %s
                                        AND connection_date < %s) op_data_sq
                           GROUP BY imei_norm, imsi_norm, msisdn_norm) triplet_sq""")
                           .format(sql.Identifier(aggregated_data_temp_table),
                                   sql.Identifier(src_partition),
                                   msisdn_output=msisdn_output),
                           [year, month, self._operator_id, start_date, end_date])
            self._distinct_triplet_count += cursor.rowcount
            base_partition = partition_utils.monthly_network_triplets_country_partition(month=month, year=year)
            dest_partition = partition_utils.imei_shard_name(base_name=base_partition,
                                                             virt_imei_range_start=virt_imei_shard_start,
                                                             virt_imei_range_end=virt_imei_shard_end)
            on_conflict_sql = sql.SQL(
                """ON CONFLICT (triplet_hash)
                     DO UPDATE
                           SET first_seen = least(target.first_seen, excluded.first_seen),
                               last_seen = greatest(target.last_seen, excluded.last_seen),
                               date_bitmask = target.date_bitmask | excluded.date_bitmask
                         WHERE (target.date_bitmask | excluded.date_bitmask) != target.date_bitmask
                """  
            )
            cursor.execute(
                sql.SQL(
                    """INSERT INTO {0} AS target(triplet_year, triplet_month, first_seen, last_seen, date_bitmask,
                                                 triplet_hash, imei_norm, imsi, msisdn, virt_imei_shard)
                            SELECT triplet_year, triplet_month, first_seen, last_seen, date_bitmask, triplet_hash,
                                   imei_norm, imsi, msisdn, virt_imei_shard
                              FROM {1}
                                   {2}
                    """
                ).format(sql.Identifier(dest_partition), sql.Identifier(aggregated_data_temp_table), on_conflict_sql))
            base_partition = partition_utils.monthly_network_triplets_per_mno_partition(month=month, year=year,
                                                                                        operator_id=self._operator_id)
            dest_partition = partition_utils.imei_shard_name(base_name=base_partition,
                                                             virt_imei_range_start=virt_imei_shard_start,
                                                             virt_imei_range_end=virt_imei_shard_end)
            cursor.execute(
                sql.SQL(
                    """INSERT INTO {0} AS target
                            SELECT *
                              FROM {1}
                                   {2}
                    """
                ).format(sql.Identifier(dest_partition), sql.Identifier(aggregated_data_temp_table), on_conflict_sql))
            inserted_or_updated_triplet_count = cursor.rowcount
            hll_partition_base_name = '{0}_{1:02d}_{2:d}'.format(self._staging_hll_sketches_tbl_name, month, year)
            hll_partition_name = partition_utils.imei_shard_name(base_name=hll_partition_base_name,
                                                                 virt_imei_range_start=virt_imei_shard_start,
                                                                 virt_imei_range_end=virt_imei_shard_end)
            cursor.execute(sql.SQL("""CREATE UNLOGGED TABLE {0} (LIKE {1}) INHERITS ({1})""")
                           .format(sql.Identifier(hll_partition_name),
                                   self._staging_hll_sketches_tbl_id))
            cursor.execute(sql.SQL("""
                INSERT INTO {0} (triplet_hll, imei_hll, imsi_hll, msisdn_hll, imei_imsis_hll, imei_msisdns_hll,
                                 imsi_msisdns_hll, creation_date, operator_id, data_date)
                     SELECT coalesce(hll_add_agg(hll_hash_text(hash_triplet(imei_norm, imsi, msisdn)::TEXT))
                                     filter(WHERE imei_norm IS NOT NULL
                                              AND imsi_norm IS NOT NULL
                                              AND msisdn_norm IS NOT NULL), hll_empty()) AS triplet_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm)), hll_empty()) AS imei_hll,
                            coalesce(hll_add_agg(hll_hash_text(imsi_norm)), hll_empty()) as imsi_hll,
                            coalesce(hll_add_agg(hll_hash_text(msisdn_norm)), hll_empty()) AS msisdn_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm||'$'||imsi_norm)), hll_empty())
                                AS imei_imsis_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm||'$'||msisdn_norm)), hll_empty())
                                AS imei_msisdns_hll,
                            coalesce(hll_add_agg(hll_hash_text(imsi_norm||'$'||msisdn_norm)), hll_empty())
                                AS imsi_msisdns_hll,
                            CURRENT_DATE AS creation_date,
                            %s AS operator_id,
                            make_date(%s, %s, date_part('day', connection_date)::INT) AS data_date
                       FROM {1}
                      WHERE connection_date >= %s
                        AND connection_date < %s
                   GROUP BY connection_date;
                """).format(sql.Identifier(hll_partition_name), 
                            sql.Identifier(src_partition)),
                           [self._operator_id, year, month, start_date, end_date])
        return inserted_or_updated_triplet_count
def _update_monthly_network_triplets ( self , month , year , src_partition , virt_imei_shard_start , virt_imei_shard_end ) : with create_db_connection ( self . _db_config ) as conn , conn . cursor ( ) as cursor : start_date , end_date = self . _date_range_for_month_year ( month , year ) aggregated_data_temp_table = '{0}_aggregated_{1:02d}_{2:d}' . format ( src_partition , month , year ) cursor . execute ( sql . SQL ( """CREATE TEMPORARY TABLE {0} (LIKE monthly_network_triplets_per_mno)""" ) . format ( sql . Identifier ( aggregated_data_temp_table ) ) ) if self . _perform_msisdn_import : msisdn_output = sql . Identifier ( 'msisdn_norm' ) else : msisdn_output = sql . SQL ( 'NULL AS msisdn_norm' ) cursor . execute ( sql . SQL ( """
                INSERT INTO {0} (triplet_year, triplet_month, first_seen, last_seen, date_bitmask,
                                 triplet_hash, imei_norm, imsi, msisdn, operator_id, virt_imei_shard)
                     SELECT %s,
                            %s,
                            triplet_sq.first_seen,
                            triplet_sq.last_seen,
                            triplet_sq.date_bitmask,
                            triplet_sq.triplet_hash,
                            triplet_sq.imei_norm,
                            triplet_sq.imsi_norm,
                            triplet_sq.msisdn_norm,
                            %s,
                            calc_virt_imei_shard(triplet_sq.imei_norm)
                       FROM (SELECT hash_triplet(imei_norm, imsi_norm, msisdn_norm) AS triplet_hash,
                                    imei_norm,
                                    imsi_norm,
                                    {msisdn_output},
                                    MIN(connection_date) AS first_seen,
                                    MAX(connection_date) AS last_seen,
                                    bit_or(1 << (date_part('day', connection_date)::INT - 1)) AS date_bitmask
                               FROM (SELECT *
                                       FROM {1}
                                      WHERE connection_date >= %s
                                        AND connection_date < %s) op_data_sq
                           GROUP BY imei_norm, imsi_norm, msisdn_norm) triplet_sq""" ) . format ( sql . Identifier ( aggregated_data_temp_table ) , sql . Identifier ( src_partition ) , msisdn_output = msisdn_output ) , [ year , month , self . _operator_id , start_date , end_date ] ) self . _distinct_triplet_count += cursor . rowcount base_partition = partition_utils . monthly_network_triplets_country_partition ( month = month , year = year ) dest_partition = partition_utils . imei_shard_name ( base_name = base_partition , virt_imei_range_start = virt_imei_shard_start , virt_imei_range_end = virt_imei_shard_end ) on_conflict_sql = sql . SQL ( """ON CONFLICT (triplet_hash)
                     DO UPDATE
                           SET first_seen = least(target.first_seen, excluded.first_seen),
                               last_seen = greatest(target.last_seen, excluded.last_seen),
                               date_bitmask = target.date_bitmask | excluded.date_bitmask
                         WHERE (target.date_bitmask | excluded.date_bitmask) != target.date_bitmask
                """ ) cursor . execute ( sql . SQL ( """INSERT INTO {0} AS target(triplet_year, triplet_month, first_seen, last_seen, date_bitmask,
                                                 triplet_hash, imei_norm, imsi, msisdn, virt_imei_shard)
                            SELECT triplet_year, triplet_month, first_seen, last_seen, date_bitmask, triplet_hash,
                                   imei_norm, imsi, msisdn, virt_imei_shard
                              FROM {1}
                                   {2}
                    """ ) . format ( sql . Identifier ( dest_partition ) , sql . Identifier ( aggregated_data_temp_table ) , on_conflict_sql ) ) base_partition = partition_utils . monthly_network_triplets_per_mno_partition ( month = month , year = year , operator_id = self . _operator_id ) dest_partition = partition_utils . imei_shard_name ( base_name = base_partition , virt_imei_range_start = virt_imei_shard_start , virt_imei_range_end = virt_imei_shard_end ) cursor . execute ( sql . SQL ( """INSERT INTO {0} AS target
                            SELECT *
                              FROM {1}
                                   {2}
                    """ ) . format ( sql . Identifier ( dest_partition ) , sql . Identifier ( aggregated_data_temp_table ) , on_conflict_sql ) ) inserted_or_updated_triplet_count = cursor . rowcount hll_partition_base_name = '{0}_{1:02d}_{2:d}' . format ( self . _staging_hll_sketches_tbl_name , month , year ) hll_partition_name = partition_utils . imei_shard_name ( base_name = hll_partition_base_name , virt_imei_range_start = virt_imei_shard_start , virt_imei_range_end = virt_imei_shard_end ) cursor . execute ( sql . SQL ( """CREATE UNLOGGED TABLE {0} (LIKE {1}) INHERITS ({1})""" ) . format ( sql . Identifier ( hll_partition_name ) , self . _staging_hll_sketches_tbl_id ) ) cursor . execute ( sql . SQL ( """
                INSERT INTO {0} (triplet_hll, imei_hll, imsi_hll, msisdn_hll, imei_imsis_hll, imei_msisdns_hll,
                                 imsi_msisdns_hll, creation_date, operator_id, data_date)
                     SELECT coalesce(hll_add_agg(hll_hash_text(hash_triplet(imei_norm, imsi, msisdn)::TEXT))
                                     filter(WHERE imei_norm IS NOT NULL
                                              AND imsi_norm IS NOT NULL
                                              AND msisdn_norm IS NOT NULL), hll_empty()) AS triplet_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm)), hll_empty()) AS imei_hll,
                            coalesce(hll_add_agg(hll_hash_text(imsi_norm)), hll_empty()) as imsi_hll,
                            coalesce(hll_add_agg(hll_hash_text(msisdn_norm)), hll_empty()) AS msisdn_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm||'$'||imsi_norm)), hll_empty())
                                AS imei_imsis_hll,
                            coalesce(hll_add_agg(hll_hash_text(imei_norm||'$'||msisdn_norm)), hll_empty())
                                AS imei_msisdns_hll,
                            coalesce(hll_add_agg(hll_hash_text(imsi_norm||'$'||msisdn_norm)), hll_empty())
                                AS imsi_msisdns_hll,
                            CURRENT_DATE AS creation_date,
                            %s AS operator_id,
                            make_date(%s, %s, date_part('day', connection_date)::INT) AS data_date
                       FROM {1}
                      WHERE connection_date >= %s
                        AND connection_date < %s
                   GROUP BY connection_date;
                """ ) . format ( sql . Identifier ( hll_partition_name ) , sql . Identifier ( src_partition ) ) , [ self . _operator_id , year , month , start_date , end_date ] ) return inserted_or_updated_triplet_count
Helper function to update the monthly_network_triplets tables (country and per-MNO).
Helper function to update the monthly_network_triplets tables ( country and per - MNO ) .
"""Helper function to update the monthly_network_triplets tables (country and per-MNO).""" # Save how many triplets actually got inserted for later on when we print stats # Need to insert into both country and per-MNO tables. Do country first # The on conflict clause is common to both insertions # noqa: Q441 # Now insert into the per-MNO table # Update daily_per_mno_hll_sketches table # noqa Q441, Q447
self
month
year
src_partition
virt_imei_shard_start
virt_imei_shard_end