Notes

def export_tables(final_table_name, is_parquet, app_name, record_count):
try:
if app_name == 'bidb':
app.logger.info('Table name processing for : %s', str(final_table_name))
final_table_details = utils.get_final_table_status(final_table_name)
final_table_status = final_table_details[0].get('final_table_status')
export_status = final_table_details[0].get('export_status')

if final_table_status.lower() == 'completed' and export_status.lower() not in ('in process', 'completed'):
bidb_bucket_name = app.config['BIDB_BUCKET_NAME']
bidb_prefix = app.config['BIDB_FINAL_DATASET_PREFIX']
utils.update_csv_value(final_table_name, 'export_status', 'In Process')
if is_parquet:
utils.update_csv_value(final_table_name, 'export_parquet_status', 'In Process')

if final_table_name == 'BIDB_Change_Detection':
thread_process = threading.Thread(target=bidb_change_detection_save_files, args=(bidb_bucket_name, bidb_prefix, is_parquet))
else:
thread_process = threading.Thread(target=process_and_save_files, args=(bidb_bucket_name, bidb_prefix, final_table_name,is_parquet, final_table_details))
thread_process.start()
app.logger.info('Started the thread')
return "in_process"
else:
app.logger.warning('The current final table is already in process or has been processed')
return "processed"
elif app_name == 'pmtx_infous':
app.logger.info('Starting process for Pmtx Infous')
thread_process = threading.Thread(target=pmtx_infous_save_csv)
thread_process.start()
app.logger.info('Started the thread')
return "in_process"

except Exception:
app.logger.info(traceback.format_exc())
def process_and_save_files(bucket_name, prefix, final_table_name, is_parquet, final_table_details):
start_time = datetime.datetime.now()
try:
prefix = prefix + final_table_name+'/'
if final_table_name == 'BIDB_Valid_NAICS_Lookup':
prefix = 'config_csvs/BIDB_Valid_NAICS_Lookup/'
elif final_table_name == 'BIDB_Naics_Ranking_Config':
prefix = 'config_csvs/BIDB_Naics_Ranking_Config/'
save_csv_from_athena_view(prefix=prefix)
app.logger.info(f"Using bucket::: {bucket_name}")
app.logger.info(f"Fetching files from prefix ::: {prefix}")
parquet_files = utils.list_s3_objects_with_pagination(bucket_name, prefix)
parquet_files = parquet_files['Contents']
app.logger.info(f"found {len(parquet_files)} files")
if len(parquet_files) >0:
chunk_size = 500_000 if app.config['ENV'] == 'prod' else 100_000
chunk_size = 100_000 if final_table_name.lower() == 'bidb_yelp' else chunk_size
current_chunk = pd.DataFrame()
record_count = 0
csv_file_count = 1
total_exported_record_count = 0
number_of_records = int(final_table_details[0]['record_count'])
expected_number_of_files = math.ceil(number_of_records/chunk_size)
file_names =[]
meta_data_headers = ["FileName", "RecordCount", "ProcessDateTime"]
meta_data_list =[]

record_id_index = 1 # Starts from 1 in v2 version
for file_key in parquet_files:
"""
This aims to create new files of with target chunk_size each. For instance we have 10 files with variable length of
data in it.

For simplicity: 1st file has 700k records.
1. It checks len(file_data) which is 700k greater than 0
2. Checks <remaing_space> = <chunk_size> 500k - <record_count> 0 ===> 500k
3. <file_data> (700k) > remaining_space (500k) so file is split into two. 1st::: <part> (with 500k records)
and 2nd::: <file_data> (with 200k records)
4. <current_chunk> gets complete 500k records and gets merged with empty dataframe defined earlier
5. Since chunk size has been reached new csv file is saved and record_count is reset to 0




1. Now in second iteration, remaining_space = chunk_size (500k) - record_count (0)
2. file_data (200k <--- check above) is less than remaining space (500k) so it goes to else part
3. <part> is whole remaing chunk with 200k
and <file_data> is made empty so it does not get processed agin in while loop.
4. <current_chunk> gets 200k concated, <record_count> = 200k
5. Since <record_count> 200k is not greater than <chunk_size> 500k so new csv is not saved.



1. At this point we have got out from while loop and start processing 2nd file.
2. <remaining_space> = <chunk_size> 500k - <record_count> 200k ===> 300k
3. From now on the process gets start repeating
"""
if not file_key['Size'] > 0:
continue

app.logger.info(f"Processing {file_key}")
if final_table_name in ('BIDB_Valid_NAICS_Lookup', 'BIDB_Naics_Ranking_Config'):
file_data = utils.read_csv_file_from_s3(bucket_name, file_key['Key'])
else:
file_data = utils.read_parquet_from_s3(bucket_name, file_key['Key'], final_table_name)

if final_table_name == 'BIDB_Change_Detection':
true_columns = [col for col, val in app.config['CHANGE_DETECTION_COLUMNS'].items() if val]
delivery_batch_id = file_key['Key'].split('/')[3]
all_columns_change_detection = file_data.columns.tolist()
missing_columns = [col for col in true_columns if col not in all_columns_change_detection]
for col in missing_columns:
file_data[col] = None
file_data = file_data[true_columns]
file_data['Delivery_BatchID'] = delivery_batch_id

if final_table_name == 'BIDB_Additional_Attributes':
file_data['attribute_name'] = utils.get_additional_attributes_name_from_partition(file_key['Key'].split('/')[-2])
while len(file_data) > 0:
remaining_space = chunk_size - record_count
if len(file_data) > remaining_space:
# Split the data: Take the remaining space in the current CSV
part = file_data.iloc[:remaining_space] # Take remaining space in current chunk
file_data = file_data.iloc[remaining_space:] # Remove processed part from data
else:
part = file_data
file_data = pd.DataFrame() # Clear file_data after processing

# Append the chunk to the combined data
current_chunk = pd.concat([current_chunk, part], ignore_index=True)
record_count += len(part)

# If we have reached the target, save to CSV and reset
if record_count >= chunk_size:
file_name,failure_message = save_csv(current_chunk, final_table_name, csv_file_count, record_id_index, is_parquet)
if 'Failed Validation' in failure_message:
app.logger.info(f"Company Overview Attribute Validation is Failed {failure_message}")
utils.update_csv_value(final_table_name, "export_status", "Failed")
utils.update_csv_value(final_table_name, "failure_message",failure_message)
if is_parquet:
utils.update_csv_value(final_table_name, 'export_parquet_status', 'Failed')
subject = f'Validation failed for {final_table_name}'
utils.send_generic_sns(final_table_name, "Failed",subject, failure_message)
break
record_id_index += len(current_chunk)
current_chunk = pd.DataFrame()
meta_data_list.append([final_table_name+"_"+str(csv_file_count),record_count, datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d')])
total_exported_record_count += record_count
record_count = 0 #reset record count
csv_file_count += 1
file_names.append(file_name)
if app.config['EXPORT_ONLY_100']:
app.logger.info("Only exporting 100k -1")
break # only export 100 record for dev and local
if app.config['EXPORT_ONLY_100']:
app.logger.info("Only exporting 100k -2")
break
if not current_chunk.empty:
total_exported_record_count += len(current_chunk)
file_name,failure_message = save_csv(current_chunk, final_table_name, csv_file_count, record_id_index, is_parquet)
if 'Failed Validation' in failure_message:
app.logger.info(f"Company Overview Attribute Validation is Failed {failure_message}")
utils.update_csv_value(final_table_name, "export_status", "Failed")
utils.update_csv_value(final_table_name, "failure_message",failure_message)
if is_parquet:
utils.update_csv_value(final_table_name, 'export_parquet_status', 'Failed')
subject = f'Validation failed for {final_table_name}'
utils.send_generic_sns(final_table_name, "Failed",subject, failure_message)
return subject
record_id_index += len(current_chunk)
meta_data_list.append([final_table_name+"_"+str(csv_file_count),record_count, datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d')])
file_names.append(file_name)

meta_data_list.append([final_table_name, total_exported_record_count, datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d')])
app.logger.info(meta_data_list)
meta_data_df = pd.DataFrame(meta_data_list, columns=meta_data_headers)
save_csv(meta_data_df, final_table_name+"_MetaData", 0, None)
utils.update_csv_value(final_table_name, "export_status", "Completed")
if is_parquet:
utils.update_csv_value(final_table_name, 'export_parquet_status', 'Completed')
end_time = datetime.datetime.now()
diff = end_time - start_time
formatted_diff = str(diff).replace(',', ':')
utils.update_csv_value(final_table_name, 'export_time_taken', str(formatted_diff))
all_files_exported = utils.check_all_parquet_files_are_exported()
app.logger.info(f"All files exported: {all_files_exported}")
if final_table_name.lower() == 'BIDB_Additional_Attributes'.lower():
check = utils.validate_additional_attributes_value_and_name(folder_name = utils.get_bucket_foldername(),
final_table_name=final_table_name)
if not check:
raise Exception("Additional Attributes Validation failed")

#TODO: Add validation if expected file count does not match
if all_files_exported:
utils.check_large_columns()
#TODO: Add validation if expected file count does not match
# if expected_number_of_files == csv_file_count:
# print("Expected file count matches")
except Exception as e:
description = str(e)
utils.update_csv_value(final_table_name, "export_status", "Failed")
if is_parquet:
utils.update_csv_value(final_table_name, 'export_parquet_status', 'Failed')
app.logger.error(f"Exception in export_table_to_csv :: {e}")
subject = f'Export failed for {final_table_name}'
utils.send_generic_sns(final_table_name, "Failed",subject, description)