"""This is the old cludgey logic for process_release_list()""" from datetime import datetime import pytz import logging import math from sys import exit as sysexit from constants import db from utils import snowflake_utils import config if not config.SFN_NAME: sysexit('You must export SFN_NAME in your environment with the SFN name.') SLEEP_DELAY = 10 # create logger with 'spam_application' logger = logging.getLogger('test') logger.setLevel(logging.DEBUG) # create console handler with a higher log level ch = logging.StreamHandler() ch.setLevel(logging.INFO) # create formatter and add it to the handlers formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') # noqa ch.setFormatter(formatter) # add the handlers to the logger logger.addHandler(ch) # Get "today" at runtime TODAY = datetime.now(pytz.utc).date().isoformat() def process_release_list(release_list, table_name): """Convert the release list into chunked tables. Given the MAX_CONCURRENT_SFN calculate a better MAX_ASSETS_PER_TABLE for optimum parallelism. Args: release_list (list): List of releases. table_name (str): Name of the table. """ release_count = 0 asset_count = 0 total_assets_seen = 0 table_num = 1 table_dict = {} upc_asset_list = [] total_asset_count = \ sum([release[db.ASSET_COUNT] for release in release_list]) # Process max rows per table if config.ASSETS_PER_TABLE: # Use the config value logger.info( f'Using config value for max assets per table: {config.ASSETS_PER_TABLE}') # noqa assets_per_table = config.ASSETS_PER_TABLE else: # Calculate based on MAX_CONCURRENT_SFN logger.info( f'Calculating assets per table based on MAX_CONCURRENT_SFN={config.MAX_CONCURRENT_SFN}') # noqa row_count = snowflake_utils.get_table_row_count(table_name=table_name) logger.info(f'Total rows in table: {row_count}') # We remove one because we want to ensure that non-uniform # distributions don't result in small row sets. By adding one, we # ensure that the distribution is eased. This will either result in the # MAX_CONCURRENT_SFN being deployed, or a better distribution across # extra tables as needed. The 500 row limit is a hard limit, so we can # be more aggressive in our calculations. assets_per_table = int(math.ceil(row_count / (config.MAX_CONCURRENT_SFN-1))) # noqa if assets_per_table > 500: assets_per_table = 500 logger.info(f'{assets_per_table} is too high. Max assets per ' 'table set to 500.') else: logger.info(f'Max asets per table: {assets_per_table}') while release_list: left_to_go = len(release_list) logger.info(f'{left_to_go} releases left to process.') assets_remaining = \ sum([release[db.ASSET_COUNT] for release in release_list]) logger.info(f'{assets_remaining} assets remaining.') midpoint = left_to_go//2 # Sort the release list by asset count in descending order low_asset_counts = list(reversed(release_list[:midpoint])) high_asset_counts = list(reversed(release_list[midpoint:])) # Debugging assert len(high_asset_counts) >= len(low_asset_counts) # This list will track the indices of the release_list that have been # used in the current table. This will be used to delete the used # releases from the main release list. used_release_list_indices = [] # Start with the highest asset counts, biggest asset counts first for i, high_asset_release in enumerate(high_asset_counts): # There is still space in table for large item if asset_count+high_asset_release[db.ASSET_COUNT] <= assets_per_table: # noqa asset_count += high_asset_release[db.ASSET_COUNT] # Increment the total asset count total_assets_seen += high_asset_release[db.ASSET_COUNT] # Increment the release count release_count += 1 # Add the upc and asset count to the upc_asset_list upc_asset_list.append(( high_asset_release[db.UPC], high_asset_release[db.ASSET_COUNT] )) # Get the index of the high asset item in the og release list high_release_list_index = \ release_list.index(high_asset_release) # Debugging assert release_list[high_release_list_index] == high_asset_release # noqa # Mark the release for deletion from the main release list used_release_list_indices.append(high_release_list_index) else: # No room for large item used_low_assets = [] # Track used low assets # Loop through the low asset counts by descending asset count for j, low_asset_release in enumerate(low_asset_counts): if asset_count+low_asset_release[db.ASSET_COUNT] <= assets_per_table: # noqa # Table asset count asset_count += low_asset_release[db.ASSET_COUNT] # Increment the total asset count total_assets_seen += low_asset_release[db.ASSET_COUNT] # Increment the release count release_count += 1 # Add the upc and asset count to the upc_asset_list upc_asset_list.append(( low_asset_release[db.UPC], low_asset_release[db.ASSET_COUNT] )) # Get index of low asset count in the og release list low_release_list_index = release_list.index(low_asset_release) # noqa # Debugging assert release_list[low_release_list_index] == low_asset_release # noqa # Remove used low assets from release_list used_release_list_indices.append(low_release_list_index) # noqa # Add used low assets to used_low_assets used_low_assets.append(low_asset_release) # Remove used low assets from low_asset_counts for used_asset in used_low_assets: low_asset_counts.remove(used_asset) # Pinch off the table, and reset the loop vars tmp_table_name = \ config.ENVIRONMENT+'_'+table_name+'_'+str(table_num).zfill(3) # noqa table_dict[tmp_table_name] = upc_asset_list logger.info(f'Adding table {tmp_table_name} ' f'with {len(upc_asset_list)} UPC\'s and ' f'{asset_count} assets.') upc_asset_list = [] table_num += 1 asset_count = 0 logger.info( f'Marking {len(used_release_list_indices)} releases as chunked.') # report assets chunked logger.info(f'{total_assets_seen} assets chunked.') logger.info(f'{total_asset_count - total_assets_seen} assets ' 'remaining.') logger.info(f'{release_count} releases chunked.') logger.info(f'{left_to_go - len(used_release_list_indices)} releases ' 'remaining.') # Remove used high assets from release_list release_list = [ v for i, v in enumerate(release_list) if i not in used_release_list_indices # noqa ] # Debugging assert len(release_list) == left_to_go - len(used_release_list_indices) # If there are any low assets left, add them to the last table if len(release_list) == 1: logger.info('Last release found. It has ' f'{release_list[0][db.ASSET_COUNT]} assets.') # Add the last release to the list upc_asset_list.append(( release_list[0][db.UPC], release_list[0][db.ASSET_COUNT] )) total_assets_seen += release_list[0][db.ASSET_COUNT] asset_count += release_list[0][db.ASSET_COUNT] release_count += 1 release_list.pop() if len(upc_asset_list): # Last table # Add environment to table names to prevent collisions tmp_table_name = \ config.ENVIRONMENT+'_'+table_name+'_'+str(table_num).zfill(3) table_dict[tmp_table_name] = upc_asset_list logger.info(f'Adding last table {tmp_table_name} ' f'with {len(upc_asset_list)} upcs and ' f'{asset_count} assets.') upc_asset_list = [] asset_count = 0 logger.info(f'{total_assets_seen} assets chunked.') logger.info(f'{release_count} releases chunked.') logger.info(f'{table_num} tables prepared for creation.') # Print out the final list of tables and their asset and upc counts for table_name, upc_list in table_dict.items(): logger.info( f'{table_name}: {len(upc_list)} upcs, ' f'{sum([upc_asset_count[1] for upc_asset_count in upc_list])} assets') # noqa return table_dict