S3 にある細切れのファイルを読み込んで別のデータベースにインポートしたいです
プレフィックスリストしてオブジェクトを1個ずつ読むというコードを、asyncio なしでかいたところ、3万ぐらいのファイルを読み込むのに40分ぐらいかかってしまい、CPUは 20% ぐらいしか使ってなかったのでおそらく S3 アクセスがボトルネックなのかなと想像しています
なので asyncio を使って S3 に4並列にアクセスする様にしたいです
バケットリスト部分を
https://pypi.org/project/aioboto3/
オブジェクト取得を
https://pypi.org/project/aiobotocore/
を参考にしてあとは動かしながら子ルーチン型ですよっていうエラーがでたら適当に await をはさんでいって
動く様にはなったんですがまったく並列化してるようにみえません
import asyncio
import aioboto3
async def get_object(s3, bucket, key):
print("start get object")
obj = await s3.Object(bucket, key)
response = await obj.get()
async with response['Body'] as stream:
gz_data = await stream.read()
print("end get object")
async def list_bucket():
### list json ###
bucket_name = 'test-bucket'
prefix = 'prefix/'
async with aioboto3.resource("s3") as s3:
bucket = await s3.Bucket(bucket_name) # <----------------
sem = asyncio.Semaphore(4)
async for s3_object in bucket.objects.filter(Prefix=prefix):
await sem.acquire()
await get_object(s3, bucket_name, s3_object.key)
sem.release()
def main():
loop = asyncio.get_event_loop()
loop.run_until_complete(list_bucket())
if __name__ == '__main__':
main()
get_object 内には 4 つまで並列に入れて
obj = await s3.Object(bucket, key)
response = await obj.get()
async with response['Body'] as stream:
gz_data = await stream.read()
ここのどこかで S3 からデータをもってくるのでブロックして処理を手放すはずなので
実行すると
start get object
start get object
start get object
start get object
end get object
:
と最初に s3 object 取得リクエストだけ4並列に発行してほしいんですが
実際に実行すると
start get object
end get object
start get object
end get object
:
という感じで get_object が直列にしか実行されてない様に見えます
thread fork join 形式の並列処理しかかいたことがなく
asyncio を初めて使うのでいまいち理解できていないんですが
await get_object(s3, bucket_name, s3_object.key)
ここに await が必要な理由が分からないんですよね
get_object の終わりは待たずにどんどん次のループに進んで
次の get_object を実行して欲しいんですが
これをかかないとそもそも get_object の中を実行してくれません
asyncio で s3 に並列アクセスするにはどうかけばいいのでしょうか
追記です
コメントで教えていただいた create_task で
s3 からオブジェクトを取得する部分は並列にできたのですが、
ページを読む部分を先に実行しないと create_task ができなくて困ってます。
ファイル数が数万個になると先に数万個のキー配列を取得してしまうのもメモリが無駄なので
ページごとにオブジェクトを読んでは読んだキーは捨てるというようにしたいです
async for s3_object in bucket.objects.filter(Prefix=prefix):
の部分をさらに分解する必要がありそうなのですが async for と通常の for は何が違うのでしょうか