如何从位于 S3 中的 130,000 多个 Json 文件中快速提取信息？

Question

我有一个 S3 超过 130k Json 文件，我需要根据 json 文件中的数据计算数字（例如计算发言人的性别数量）。我目前正在使用 s3 Paginator 和 JSON.load 来读取每个文件并提取信息表单。但是处理如此大量的文件（每秒2-3个文件）需要很长时间。我怎样才能加快这个过程？如果可能，请提供工作代码示例。谢谢 这是我的一些代码：

client = boto3.client('s3')
paginator = client.get_paginator('list_objects_v2')
result = paginator.paginate(Bucket='bucket-name',StartAfter='') 

for page in result:
    if "Contents" in page:
        for key in page[ "Contents" ]:
            keyString = key[ "Key" ]
                s3 = boto3.resource('s3')
                content_object = s3.Bucket('bucket-name').Object(str(keyString))
                    file_content = content_object.get()['Body'].read().decode('utf-8')
                    json_content = json.loads(file_content)
                    x = (json_content['dict-name'])

Answer 1

为了使用下面的代码，我假设您了解 pandas（如果不了解，您可能想了解它）。此外，尚不清楚您的 2-3 秒是在读取还是包含部分数字运算，但多处理将大大加快这一速度。要点是读取（作为数据帧）中的所有文件，将它们连接起来，然后进行分析。

为了对我有用，我在具有大量 vCPU 和内存的现场实例上运行它。例如，我发现网络优化的实例（如 c5n - 查找 n）和 inf1（用于机器学习）在读/写方面比 T 或 M 实例类型快得多。

我的用例是读取 2000 个“目录”，每个目录大约有 1200 个文件并分析它们。多线程比单线程快几个数量级。

文件 1：您的主脚本

# create script.py file
import os
from multiprocessing import Pool
from itertools import repeat
import pandas as pd
import json
from utils_file_handling import *

ufh = file_utilities() #instantiate the class functions - see below (second file)

bucket = 'your-bucket'
prefix = 'your-prefix/here/' # if you don't have a prefix pass '' (empty string or function will fail)

#define multiprocessing function - get to know this to use multiple processors to read files simultaneously
def get_dflist_multiprocess(keys_list, num_proc=4):
    with Pool(num_proc) as pool:
        df_list = pool.starmap(ufh.reader_json, zip(repeat(bucket), keys_list), 15)
        pool.close()
        pool.join()
    return df_list

#create your master keys list upfront; you can loop through all or slice the list to test
keys_list = ufh.get_keys_from_prefix(bucket, prefix)
# keys_list = keys_list[0:2000] # as an exampmle

num_proc = os.cpu_count() #tells you how many processors your machine has; function above defaults to 4 unelss given
df_list = get_dflist_multiprocess(keys_list, num_proc=num_proc) #collect dataframes for each file
df_new = pd.concat(df_list, sort=False) 
df_new = df_new.reset_index(drop=True)
# do your analysis on the dataframe 

文件 2：类函数

#utils_file_handling.py
# create this in a separate file; name as you wish but change the import in the script.py file
import boto3
import json
import pandas as pd   

#define client and resource
s3sr = boto3.resource('s3')
s3sc = boto3.client('s3')

class file_utilities:
    """file handling function"""

    def get_keys_from_prefix(self, bucket, prefix):
        '''gets list of keys and dates for given bucket and prefix'''
        keys_list = []
        paginator = s3sr.meta.client.get_paginator('list_objects_v2')
        # use Delimiter to limit search to that level of hierarchy
        for page in paginator.paginate(Bucket=bucket, Prefix=prefix, Delimiter='/'):
            keys = [content['Key'] for content in page.get('Contents')]
            print('keys in page: ', len(keys))
            keys_list.extend(keys)
        return keys_list

    def read_json_file_from_s3(self, bucket, key):
        """read json file"""
        bucket_obj = boto3.resource('s3').Bucket(bucket)
        obj = boto3.client('s3').get_object(Bucket=bucket, Key=key)
        data = obj['Body'].read().decode('utf-8')
        return data

    # you may need to tweak this for your ['dict-name'] example; I think I have it correct
    def reader_json(self, bucket, key):
        '''returns dataframe'''
        return pd.DataFrame(json.loads(self.read_json_file_from_s3(bucket, key))['dict-name'])