如何有效地枚举 AWS S3 密钥空间

Question

我正在构建一个应用程序，它需要列出存储在 AWS S3 中的大量对象（比如说 500M 到 10 亿个对象）。直接通过分页列出对象需要数周时间。我想并行化列表，但要有效地做到这一点，我需要映射基本上未知的键空间。

有关更多背景信息，AWS 允许您提供前缀和分隔符作为 ListBuckets 操作的一部分。见这里：http://docs.aws.amazon.com/AmazonS3/latest/dev/ListingKeysHierarchy.html

所以，看起来我需要一种算法，可以枚举未知的 S3 密钥空间（前缀空间？）并尝试将发现的前缀分组到大约 [X] 个大小相等的桶中，这些桶可以并行化以实现均匀采样和速度列表。

高级伪代码：

1. 使用前缀为 [0-9a-z] 的 N（比如 3 个）字母组合和 1k 的最大键大小来执行并行搜索。
2. 对于返回超过 max-keys 键（我们不知道确切大小）的任何搜索，使用 [discovered prefix]+[0-9a-z] 前缀执行后续 GET。如果搜索返回 1-999 个键，则添加到单独的存储桶列表
3. 一旦我们对尽可能均匀地分配密钥（可能在 N 个递归步骤之后）感到满意后，任务工作人员开始列出他们自己的存储桶

挑战：

• 构建一个均匀的抽样分布以发送给工作人员列出。
• 需要尽量减少一名工人需要完成所有工作的机会。
• 如果所有前缀都以“aaaaaaaaaaaaaa”开头怎么办？ =)

目标是以编程方式发现和分组前缀，以实现对未知前缀结构列表的均匀采样和并行化。任何关于算法、链接、示例的想法都将不胜感激！

Answer 1

我创建了一个工具，可以使用生产者/消费者系统递归分析 S3 键空间，这样每个新发现的前缀都会在不同的线程中枚举。这是我能找到的最有效的方法。

如果有一种方法可以通过 ETag 检索对象，那就太好了，但是，如果 etag 在存储桶中是唯一的，那也太好了。

代码如下：

def search_objects(bucket, prefix=None, *, name, delimiter='/', limit=None, searchFoundPrefixes=True, threads=20):
    """Search for occurences of a name. Returns a list of all found keys as dictionaries.
    @param bucket - the bucket to search
    @param prefix - the prefix to start with
    @param name   - the name being searched for
    @param delimiter - the delimiter that separates names
    @param limit  - the maximum number of names keys to return
    @param searchFoundPrefixes - If true, do not search for prefixes below where name is found.
    @param threads - the number of Python threds to use. Note that this is all in the same process.
    """

    import queue
    import threading

    if limit is None:
        limit = sys.maxsize  # should be big enough
    ret = []

    def worker():
        while True:
            prefix = q.get()
            if prefix is None:
                break
            found_prefixes = []
            found_names = 0
            for obj in list_objects(bucket, prefix=prefix, delimiter=delimiter):
                if _Prefix in obj:
                    found_prefixes.append(obj[_Prefix])
                if (_Key in obj) and obj[_Key].split(delimiter)[-1] == name:
                    if len(ret) < limit:
                        ret.append(obj)
                if len(ret) > limit:
                    break
            if found_names == 0 or searchFoundPrefixes:
                if len(ret) < limit:
                    for lp in found_prefixes:
                        q.put(lp)
            q.task_done()

    q = queue.Queue()
    thread_pool = []
    for i in range(threads):
        t = threading.Thread(target=worker)
        t.start()
        thread_pool.append(t)
    q.put(prefix)

    # block until all tasks are done
    q.join()

    # stop workers
    for i in range(threads):
        q.put(None)
    for t in thread_pool:
        t.join()
    return ret