Блог

Я пытаюсь извлечь и загрузить около 50 zip-файлов из корзины s3 в базу данных Oracle с помощью Python. Каждый файл размером около 80 МБ заархивирован и содержит один csv (400 МБ несжатого csv). Процесс завершается неудачно после загрузки около 9 файлов с сообщением «Убит». Код выполняется на экземпляре Linux EC2 и выглядит следующим образом.

     # This function gets list of files in s3 bucket
    s3_files_list = get_files_from_s3(p_bucket_object=bucket_object, p_prefix=file_prefix)
    if s3_files_list:
        try:
            files_exist = True

            #Creating database 
            connection, cursor = db_oracle_connection() 
            # Start Processing files
            for source_file in s3_files_list:
                source_file_name = source_file.split('/')[-1]
                logging.info(f"Loading file {source_file_name}.")
                zip_obj = bucket_object.Object(bucket_name=BUCKET_NAME, key=source_file)
                buffer = BytesIO(zip_obj.get()["Body"].read())
                zip_file = ZipFile(buffer)

                for filename in zip_file.namelist():
                    with zip_file.open(filename, 'r') as extract_file:
                        logging.info(f"Writing file {filename}")
                        file_reader = DictReader(TextIOWrapper(extract_file, 'utf-8'), delimiter=',', quotechar='"')
                        if prefixes == TPT_PREFIX:
                            file_reader.fieldnames.append('PTIPS_FILE_NAME')

                        # store no of total columns in csv
                        file_col_count = len(file_reader.fieldnames)

                        insert_statement = f"Insert into {target_table} ({','.join(x for x in file_reader.fieldnames)}) " 
                                           f"values ({','.join(':'   str(k) for k in range(1, file_col_count   1))})"
                        logging.info(f"Inserting data into {target_table}")

                        # store no of total rows in csv
                        file_row_count = 0
                        files_data = []
                        try:
                            for line in file_reader:
                                if prefixes == TPT_PREFIX:
                                    line['PTIPS_FILE_NAME'] = source_file_name
                                files_data.append(tuple(line.values()))
                                file_row_count  = 1
                                #BATCH_SIZE = 10000
                                if len(files_data) % BATCH_SIZE == 0:
                                    cursor.executemany(insert_statement, files_data)
                                    files_data = []
                            if files_data:
                                cursor.executemany(insert_statement, files_data)
                            logging.info(f"Total row count {file_row_count}")

                        except Exception as error_load:
                            connection.rollback()                                
                        else:
                            connection.commit()
                zip_file.close()
 

Не мог бы кто-нибудь, пожалуйста, помочь решить эту проблему.