

S3 にある細切れのファイルを読み込んで別のデータベースにインポートしたいです

プレフィックスリストしてオブジェクトを１個ずつ読むというコードを、asyncio
なしでかいたところ、3万ぐらいのファイルを読み込むのに４０分ぐらいかかってしまい、CPUは 20% ぐらいしか使ってなかったのでおそらく S3
アクセスがボトルネックなのかなと想像しています

なので asyncio を使って S3 に４並列にアクセスする様にしたいです

バケットリスト部分を  
<https://pypi.org/project/aioboto3/>

オブジェクト取得を  
<https://pypi.org/project/aiobotocore/>

を参考にしてあとは動かしながら子ルーチン型ですよっていうエラーがでたら適当に await をはさんでいって  
動く様にはなったんですがまったく並列化してるようにみえません

[code]

    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()
    
[/code]

get_object 内には 4 つまで並列に入れて

[code]

     obj = await s3.Object(bucket, key)
      response = await obj.get()
      async with response['Body'] as stream:
        gz_data = await stream.read()
    
[/code]

ここのどこかで S3 からデータをもってくるのでブロックして処理を手放すはずなので  
実行すると

[code]

    start get object
    start get object
    start get object
    start get object
    end get object
      :
    
[/code]

と最初に s3 object 取得リクエストだけ４並列に発行してほしいんですが  
実際に実行すると

[code]

    start get object
    end get object
    start get object
    end get object
      :
    
[/code]

という感じで 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 は何が違うのでしょうか


CC BY-SA 4.0
2021-02-19T14:46:55.140
74141
2021-03-04T04:22:45.890
2021-03-04T04:22:45.890
question
python
asyncio を使った s3 並列アクセス


asyncio.create_task()を使うのはどうでしょうか  
<https://docs.python.org/ja/3/library/asyncio-task.html>


CC BY-SA 4.0
2021-02-20T00:50:01.947
74146
2021-02-20T00:50:01.947
28626
74141
answer