GCS composeで32以上のオブジェクトをまとめる方法
先日の記事にて パーティション化されたCSVオブジェクトをCloudSQLにimportする方法 を紹介しました。 SDKを利用して32より多いオブジェクトをまとめる場合、GCPのコミュニティのチュートリアルのコードをそのまま使っていました。（2023.8.10にアーカイブされ済み） https://github.com/GoogleCloudPlatform/community/blob/master/archived/cloud-storage-infinite-compose/index.md 文章最後に書いてあるように This code is offered for demonstration purposes only, and should not be considered production-ready. Applying it to your production workloads is up to you! コードはデモ用途のため、そのまま本番環境で使うのはで推奨しないです。先日この文を見逃して自分の環境にデプロイし、平常運転1ヶ月後、パーティション化されたCSVオブジェクトの数が増えたらバグが出ました。 事象 起きていた事象は、composeによって作られた中間オブジェクトが消されず残り続け、最終的に数TBのとてつもなく大きいオブジェクトが作成されてしまいました。一つのオブジェクトは5TiBまでというGCSの上限を超えてしまうため、処理が失敗しました。 どこがバグ 問題は関数compose_and_cleanupから呼び出している関数delete_objects_concurrentにあります。オブジェクトをまとめた後、毎回中間オブジェクトを削除していますが、そのdelete処理自体が非同期処理ですべての処理が終了するのを待たずに次のblob.composeの実行が始まります。 The delete_objects_concurrent function is very simple, using fire-and-forget delete tasks in an executor. A more robust implementation might check the futures from the submitted tasks. チュートリアルの中にちゃんと書いてあります。（もっとロバスト性のある実装は、submit済みのタスクのfuturesをチェックするとのこと） まとめようとするオブジェクトの数が少ない場合はほとんど問題がないですが、1000オブジェクトあたりを超えるとdeleteがcompose処理より遅くなるため、next_chunkが永遠に存在しているままwhileループから脱出できない状態になります。 delete処理自体は特に制限ないようですが、数百のオブジェクトを削除する場合時間かかるとドキュメントに記載されています。 https://cloud.google.com/storage/docs/deleting-objects#delete-objects-in-bulk 解決方法 最初に試したことは、記事に書いてあるようにsubmit済みのタスクのfuturesをwait処理でチェックします。つまりすべてのdelete処理が終わるまでに、compose処理を行いません。 （python並行処理の詳細は公式ドキュメントまたは他の記事をご覧ください） from concurrent.futures import ALL_COMPLETED, ThreadPoolExecutor, wait def delete_objects_concurrent(blobs, executor, client) -> None: """Delete Cloud Storage objects concurrently. Args: blobs (List[storage.Blob]): The objects to delete. executor (Executor): An executor to schedule the deletions in. client (storage.Client): Cloud Storage client to use. """ futures = [] for blob in blobs: logger.info("Deleting slice {}".format(blob.name)) futures.append(executor.submit(blob.delete, client=client)) wait(futures, return_when=ALL_COMPLETED) logger.info(f"Deleted {len(blobs)} objects") しかし大量なオブジェクトを1つずつ削除していること自体は変わらないので、試してみたら処理が途中でtimeoutになりました。 batch を利用して削除処理をバッチにまとめる方法もありますが、ロジックをシンプルにしたいので、他の方法を考えました。 ...