Наиболее эффективный способ пакетного удаления файлов S3

Я хотел бы иметь возможность пакетно удалять тысячи или десятки тысяч файлов одновременно в S3. Каждый файл размером от 1 МБ до 50 МБ. Естественно, я не хочу, чтобы пользователь (или мой сервер) ждали, пока файлы находятся в процессе удаления. Поэтому, вопросы:

1. Как S3 обрабатывает удаление файлов, особенно при удалении большого количества файлов?
2. Существует ли эффективный способ сделать это и заставить AWS выполнить большую часть работы? Под эффективностью я подразумеваю сделать наименьшее количество запросов в S3 и потратить наименьшее количество времени, используя наименьшее количество ресурсов на моих серверах.

AWS поддерживает массовое удаление до 1000 объектов за запрос, используя S3 REST API и различные его обертки. Этот метод предполагает, что вы знаете ключи объектов S3, которые хотите удалить (т.е. он не предназначен для обработки таких вещей, как политика сохранения, файлы, превышающие определенный размер, и т.д.).

S3 REST API может указать до 1000 файлов для удаления в одном запросе, что гораздо быстрее, чем делать индивидуальные запросы. Помните, каждый запрос — это HTTP (следовательно, TCP) запрос. Таким образом, каждый запрос несет накладные расходы. Вам просто нужно знать ключи объектов и создать HTTP-запрос (или использовать обертку на выбранном вами языке). AWS предоставляет отличную информацию об этой функции и ее использовании. Просто выберите метод, который вам больше всего подходит!

Предполагаю, что в вашем случае конечные пользователи указывают число конкретных файлов для удаления одновременно. Вместо запуска задачи, такой как “удалить все объекты, относящиеся к изображениям” или “удалить все файлы старше определенной даты” (что, полагаю, можно легко настроить отдельно в S3).

Если это так, вы будете знать ключи, которые вам нужно удалить. Это также означает, что пользователю потребуется более оперативная обратная связь о том, был ли файл успешно удален или нет. Ссылки на точные ключи должны быть очень быстрыми, так как S3 был разработан для эффективного масштабирования, несмотря на обработку крайне большого объема данных.

Если нет, вы можете изучить асинхронные API-вызовы. Вы можете немного почитать о том, как они работают в общем, из этого блога или найти информацию о том, как это сделать на вашем языке программирования. Это позволит запросу на удаление занимать свой собственный поток, и остальной код может выполняться без ожидания пользователем. Или, вы можете передать запрос в очередь… Но оба этих варианта излишне усложняют либо ваш код (асинхронный код может быть раздражающим), либо вашу среду (вам потребуется служба/демон/контейнер/сервер для обработки очереди. Поэтому я бы избегал этого сценария, если это возможно.

Изменение: У меня нет репутации для публикации более 2 ссылок. Но вы можете просмотреть комментарии Amazon о скорости запросов и производительности здесь: http://docs.aws.amazon.com/AmazonS3/latest/dev/request-rate-perf-considerations.html И комментарии FAQ S3 о том, что массовое удаление — это лучший способ, если возможно.

Крайне медленный вариант — это s3 rm --recursive, если вам действительно нравится ожидание.

Запуск параллельных s3 rm --recursive с разными шаблонами --include немного быстрее, но все равно много времени тратится на ожидание, так как каждый процесс индивидуально извлекает весь список ключей, чтобы локально выполнить сопоставление шаблонов --include.

Вход в массовое удаление.

Я обнаружил, что могу получить наибольшую скорость, удаляя 1000 ключей за раз, используя aws s3api delete-objects.

Вот пример:

cat file-of-keys | xargs -P8 -n1000 bash -c 'aws s3api delete-objects --bucket MY_BUCKET_NAME --delete "Objects=[$(printf "{Key=%s}," "$@")],Quiet=true"' _

• Опция -P8 на xargs управляет параллелизмом. В этом случае она равна 8, что означает 8 экземпляров по 1000 удалений за раз.
• Опция -n1000 говорит xargs объединять 1000 ключей для каждого вызова aws s3api delete-objects.
• Удаление ,Quiet=true или изменение его на false выведет серверные ответы.
• Примечание: В конце этой командной строки есть легко пропускаемый _. Пользователь @VladNikiforov оставил отличный комментарий о том, для чего это нужно, поэтому я просто оставлю на это ссылку.

Но как получить file-of-keys?

Если у вас уже есть список ключей, хорошо. Задача выполнена.

Если нет, вот один из способов, я предположу:

aws s3 ls "s3://MY_BUCKET_NAME/SOME_SUB_DIR" | sed -nre "s|[0-9-]+ [0-9:]+ +[0-9]+ |SOME_SUB_DIR|p" >file-of-keys

Удобная хитрость — использовать правила жизненного цикла для обработки удаления за вас. Вы можете поставить в очередь правило для удаления префикса или объектов, которые хотите, и Amazon просто позаботится об удалении.

https://docs.aws.amazon.com/AmazonS3/latest/user-guide/create-lifecycle.html

Однострочное пакетное удаление с S3API

Вот мой однострочник, основанный на других ответах на этот пост.

1. Получает пакет из 1000 ключей объектов S3 (без необходимости сохранять их в файл)
2. Передает ключи в команду удаления
3. Две команды удаления запускаются параллельно для 500 объектов каждая

aws s3api list-objects-v2 --bucket $BUCKET --prefix $PREFIX --output text --query \
'Contents[].[Key]' | grep -v -e "'" | tr '\n' '\0' | xargs -0 -P2 -n500 bash -c \
'aws s3api delete-objects --bucket $BUCKET --delete "Objects=[$(printf "{Key=%q}," "$@")],Quiet=true"' _ 

Почему без файла ключей?

Одной из жалоб на передачу ключей в файл было то, что могут возникнуть ошибки при удалении из s3. Если вам придется перезапустить команду удаления, у вас будет файл с множеством ключей, которые уже удалены, и вы потратите время на повторный запуск команд удаления.

Почему 2 команды удаления?

Я пытался удалить все 1000 объектов с помощью 1 команды удаления. Я получал ошибку, что список моих аргументов был слишком длинным (потому что у меня длинные ключи)

Меня разочаровала производительность веб-консоли для этой задачи. Я обнаружил, что команда AWS CLI хорошо справляется с этим. Например:

aws s3 rm --recursive s3://my-bucket-name/huge-directory-full-of-files

Для большой иерархии файлов это может занять значительное количество времени. Вы можете запустить это в сессии tmux или screen и проверить позже.

Уже было упоминание о команде s3 sync, но без примера и слова о параметре --delete.

Я нашел самый быстрый способ удалить содержимое папки в корзине S3 my_bucket следующим образом:

aws s3 sync --delete "local-empty-dir/" "s3://my_bucket/path-to-clear"

Я знаю, что этому посту уже очень много лет, но если вам нужно сделать это сегодня, в дашборде AWS теперь есть функция “Очистить” на странице поиска корзины, которая выполнит массовое удаление (по 1000 за раз) для вас:

Существует способ на Python, который использует botos’3 delete_objects, tqdm для предоставления небольшой обратной связи и несколько потоков для добавления небольшого параллелизма (несмотря на GIL, несколько запросов к S3 должны выполняться одновременно)

Он расположен на https://gist.github.com/michalc/20b79c9028c342ef7c38df8693f8715b, но чтобы скопировать его сюда:

# From some light testing deletes at between 2000 to 6000 objects per second, and works best if you
# have objects distributed into folders/CommonPrefixes as specified by the delimiter.

from concurrent.futures import FIRST_EXCEPTION, ThreadPoolExecutor, wait
from dataclasses import dataclass, field
from functools import partial
from typing import Callable, Optional, Tuple
from queue import PriorityQueue

import boto3
from botocore.config import Config


def bulk_delete(
    bucket, prefix,
    workers=4, page_size=1000, delimiter="https://serverfault.com/",
    get_s3_client=lambda: boto3.client('s3', config=Config(retries={'max_attempts': 16, 'mode': 'adaptive'})),
    on_delete=lambda num: None,
):
    s3 = get_s3_client()
    queue = PriorityQueue()

    @dataclass(order=True)
    class Work:
        # A tuple that determines the priority of the bit of work in "func". This is a sort of
        # "coordinate" in the paginated node tree that prioritises a depth-first search.
        priority: Tuple[Tuple[int,int], ...]
        # The work function itself that fetches a page of Key/CommonPrefixes, or deletes
        func: Optional[Callable[[], None]] = field(compare=False)

    # A sentinal "stop" Work instance with priority chosen to be before all work. So when it's
    # queued the workers will stop at the very next opportunity
    stop = Work(((-1,-1),), None)

    def do_work():
        while (work := queue.get()) is not stop:
            work.func()
            queue.task_done()
            with queue.mutex:
                unfinished_tasks = queue.unfinished_tasks
            if unfinished_tasks == 0:
                for _ in range(0, workers):
                    queue.put(stop)

    def list_iter(prefix):
        return iter(s3.get_paginator('list_objects_v2').paginate(
            Bucket=bucket, Prefix=prefix,
            Delimiter=delimiter, MaxKeys=page_size, PaginationConfig={'PageSize': page_size},
        ))

    def delete_page(page):
        s3.delete_objects(Bucket=bucket, Delete={'Objects': page})
        on_delete(len(page))

    def process_page(page_iter, priority):
        try:
            page = next(page_iter)
        except StopIteration:
            return

        # Deleting a page is done at the same priority as this function. It will often be done
        # straight after this call because this call must have been the highest priority for it to
        # have run, but there could be unfinished nodes earlier in the depth-first search that have
        # since submitted work, and so would be prioritised over the deletion
        if contents := page.get('Contents'):
            delete_priority = priority
            queue.put(Work(
                priority=delete_priority,
                func=partial(delete_page, [{'Key': obj['Key']} for obj in contents]),
            ))

        # Processing child prefixes are done after deletion and in order. Importantly anything each
        # child prefix itself enqueues should be done before the work of any later child prefixes
        # to make it a depth-first search. Appending the index of the child prefix to the priority
        # tuple of this function does this, because the work inside each child prefix will only
        # ever enqueue work at its priority or greater, but always less than the priority of
        # subsequent child prefixes or the next page.
        for prefix_index, obj in enumerate(page.get('CommonPrefixes', [])):
            prefix_priority = priority + ((prefix_index,0),)
            queue.put(Work(
                priority=prefix_priority,
                func=partial(process_page,
                             page_iter=list_iter(obj['Prefix']), priority=prefix_priority)
            ))

        # The next page in pagination for this prefix is processed after delete for this page,
        # after all the child prefixes are processed, and after anything the child prefixes
        # themselves enqueue.
        next_page_priority = priority[:-1] + ((priority[-1][0], priority[-1][1] + 1),)
        queue.put(Work(
            priority=next_page_priority,
            func=partial(process_page, page_iter=page_iter, priority=next_page_priority),
        ))

    with ThreadPoolExecutor(max_workers=workers) as worker_pool:
        # Bootstrap with the first page
        priority = ((0,0),)
        queue.put(Work(
            priority=priority,
            func=partial(process_page, page_iter=list_iter(prefix), priority=priority),
        ))

        try:
            # Run workers, waiting for the first exception, if any, raised by them
            done, _ = wait(
                tuple(worker_pool.submit(do_work) for _ in range(0, workers)),
                return_when=FIRST_EXCEPTION,
            )
        finally:
            # The workers might have all stopped cleanly, or just one since it raised an exception,
            # or none of them have stopped because the above raised an exception (e.g. on 
            # KeyboardInterrupt). By putting enough stop sentinels in the queue, we can make sure
            # to eventually stop any remaining workers in any of these cases.
            for _ in range(0, workers):
                queue.put(stop)

        # If an exception raised by a worker, re-raise it
        if e := next(iter(done)).exception():
            raise e from None

используется как:

from tqdm import tqdm

print('Удаление...')
with tqdm(unit=" objects") as pbar:
    bulk_delete(
        bucket="my-bucket", prefix='my-prefix/',
        on_delete=lambda num: pbar.update(num),
    )
print('Готово')

Как указано в gist, кажется, что удаление происходит со скоростью от 2000 до 6000 объектов в секунду, что по моему тестированию лучше, чем просто delete_objects без нескольких потоков. Но результаты могут варьироваться.

Он работает, проходя по дереву CommonPrefixes (структуры-папки в S3) и постранично каталогизируя объекты в них, но проходя и удаляя несколько ветвей дерева одновременно. Таким образом, он должен предоставить производительность лучше, чем простая пагинация + delete_objects без потоков, если существует хорошее количество таких CommonPrefixes и хорошее количество объектов на CommonPrefix.

Я нашел, что rclone довольно быстрый, поскольку использует API S3.

https://rclone.org/

rclone delete --progress --transfers=1000 <rclone_confg>:<s3_bucket_and_prefix>

Без знания того, как вы управляете s3 корзинами, это может быть более или менее полезно.

Инструменты AWS CLI имеют опцию, называемую “sync”, которая может быть особенно эффективной, чтобы гарантировать, что S3 имеет правильные объекты. Если вы или ваши пользователи управляете S3 с локальной файловой системы, вы можете сэкономить массу работы по определению, какие объекты нужно удалить, используя CLI.

http://docs.aws.amazon.com/cli/latest/reference/s3/sync.html

Я написал скрипт на Python для этого.

P.S. он уничтожает ваш аккаунт s3, все корзины.

import concurrent.futures
import boto3

def purge_bucket(Bucket, S3Client):
    response = S3Client.list_objects_v2(Bucket=Bucket)
    while 'Contents' in response and response['KeyCount'] > 0:
        for key in response['Contents']:
            value = key['Key']
            key.clear()
            key['Key'] = value
        print(f'Удаление {len(response["Contents"])} ключей в {Bucket}')
        out = S3Client.delete_objects(
            Bucket=Bucket, 
            Delete={'Objects': response['Contents']}
        )
        if 'Errors' in out:
            print(f'Ошибки в {Bucket}: {out["Errors"]}')
        response = S3Client.list_objects_v2(Bucket=Bucket)
    return Bucket

s3 = boto3.client('s3')
response = s3.list_buckets()
if len(response['Buckets']) > 0:
    with concurrent.futures.ThreadPoolExecutor() as executor:
        runs = []
        for bucket in response['Buckets']:
            bucket = bucket['Name']
            runs.append(executor.submit(purge_bucket, Bucket=bucket, S3Client=s3))
        for run in concurrent.futures.as_completed(runs):
            try:
                end = s3.delete_bucket(Bucket=run.result())
                print(end)
            except Exception as e:
                print(f'{run.result()}: {e}')