多处理池中的 Boto3 客户端因“botocore.exceptions.NoCredentialsError：无法找到凭据”而失败

Question

我正在使用 boto3 连接到 s3，下载对象并进行一些处理。我正在使用多处理池来执行上述操作。

以下是我正在使用的代码的概要：

session = None

def set_global_session():
    global session
    if not session:
        session = boto3.Session(region_name='us-east-1')

def function_to_be_sent_to_mp_pool():
    s3 = session.client('s3', region_name='us-east-1')
    list_of_b_n_o = list_of_buckets_and_objects
    for bucket, object in list_of_b_n_o:
        content = s3.get_object(Bucket=bucket, Key=key)
        data = json.loads(content['Body'].read().decode('utf-8'))
        write_processed_data_to_a_location()

def main():
    pool = mp.Pool(initializer=set_global_session, processes=40)
    pool.starmap(function_to_be_sent_to_mp_pool, list_of_b_n_o_i)

现在，当processes=40 时，一切正常。 processes = 64的时候，还是不错的。

但是，当我增加到 processes=128 时，我收到以下错误：

botocore.exceptions.NoCredentialsError: Unable to locate credentials

我们的机器具有访问 S3 所需的 IAM 角色。此外，发生的奇怪事情是，对于某些进程，它可以正常工作，而对于其他一些进程，它会引发凭据错误。为什么会发生这种情况，如何解决？

发生的另一件奇怪的事情是，我能够在 2 个单独的终端选项卡中触发两个作业（每个选项卡都有一个单独的 ssh 登录 shell）。每个作业产生 64 个进程，而且运行良好，这意味着有 128 个进程同时运行。但是一个登录 shell 中的 80 个进程失败了。

跟进：

我尝试以一种方法为单独的进程创建单独的会话。另一方面，我使用boto3.client 直接创建了s3-client。但是，它们都抛出了 80 个进程的相同错误。

我还使用以下额外配置创建了单独的客户端：

Config(retries=dict(max_attempts=40), max_pool_connections=800)

这允许我一次使用 80 个进程，但任何大于 80 的进程都会失败并出现相同的错误。

后跟进：

有人可以确认他们是否能够在具有 128 个进程的多处理中使用 boto3 吗？

Answer 1

这实际上是获取凭据的竞争条件。我不确定如何在后台获取凭据，但我在 Stack Overflow 中看到了 this question，在 github 中看到了 this ticket。

我能够通过为每个进程保持随机等待时间来解决此问题。以下是适用于我的更新代码：

client_config = Config(retries=dict(max_attempts=400), max_pool_connections=800)
time.sleep(random.randint(0, num_processes*10)/1000) # random sleep time in milliseconds
s3 = boto3.client('s3', region_name='us-east-1', config=client_config)

我尝试保持睡眠时间的范围小于 num_processes*10，但同样的问题再次失败。

@DenisDmitriev，由于您正在获取凭据并明确存储它们，我认为这解决了竞争条件，因此问题得到了解决。

PS：max_attempts 和 max_pool_connections 的值没有逻辑。在弄清竞态条件之前，我插入了几个值。

Answer 2

我怀疑 AWS 最近降低了元数据请求的限制，因为我突然开始遇到同样的问题。似乎可行的解决方案是在创建池之前查询一次凭据，并让池中的进程显式使用它们，而不是让它们再次查询凭据。

我将 fsspec 与 s3fs 一起使用，我的代码如下所示：

def get_aws_credentials():
    '''
    Retrieve current AWS credentials.
    '''
    import asyncio, s3fs
    fs = s3fs.S3FileSystem()

    # Try getting credentials
    num_attempts = 5
    for attempt in range(num_attempts):
        credentials = asyncio.run(fs.session.get_credentials())
        if credentials is not None:
            if attempt > 0:
                log.info('received credentials on attempt %s', 1 + attempt)
            return asyncio.run(credentials.get_frozen_credentials())

        time.sleep(15 * (random.random() + 0.5))

    raise RuntimeError('failed to request AWS credentials '
                       'after %d attempts' % num_attempts)


def process_parallel(fn_d, max_processes):
    # [...]
    c = get_aws_credentials()

    # Cache credentials
    import fsspec.config
    prev_s3_cfg = fsspec.config.conf.get('s3', {})
    try:
        fsspec.config.conf['s3'] = dict(prev_s3_cfg,
                                        key=c.access_key,
                                        secret=c.secret_key)

        num_processes = min(len(fn_d), max_processes)

        from concurrent.futures import ProcessPoolExecutor
        with ProcessPoolExecutor(max_workers=num_processes) as pool:
            for data in pool.map(process_file, fn_d, chunksize=10):
                yield data
    finally:
        fsspec.config.conf['s3'] = prev_s3_cfg

原始 boto3 代码看起来基本相同，除了整个 fs.session 和 asyncio.run() 歌曲和舞蹈，您将使用 boto3.Session 本身并调用它的 get_credentials() 和 get_frozen_credentials() 方法直接。