How to load files parallely from Azure Blob Storage?

Shadab Hussain 0
Viewed 16

We all have already pre-defined code blocks to read files from Azure blob storage, but at certain times due to the high number of files to be read, run time increases. In the below solution, I have tried parallelizing the task of reading files from Azure Blob Storage.

azure-blob-storage multithreading parallel-processing azure python
1 Answers

In this solution, we will be using ThreadPool. A thread pool object manages a set of worker threads to which jobs can be assigned. ThreadPool instances have the same interface as Pool instances, and their resources must be appropriately managed, either by using the pool as a context manager or manually executing close() and terminate(). It is a collection of threads assigned to perform uniform tasks. The advantage of using a thread pool pattern is defining how many threads are allowed to execute simultaneously. It is to avoid the server crashing due to high CPU load or out-of-memory conditions, e.g., the server's hardware capacity can only support up to 100 requests per second.

Before parallelizing the read file task, we need to set the AZURE_STORAGE_CONNECTION_STRING which can be obtained from Azure Portal -> Go to your storage -> Settings -> Access keys and then you will get the connection string there.

# Import Packages
import os
from multiprocessing.pool import ThreadPool as Pool
from azure.storage.blob import BlobServiceClient, BlobClient, ContainerClient, __version__

connect_str = os.getenv('AZURE_STORAGE_CONNECTION_STRING')
blob_service_client = BlobServiceClient.from_connection_string(connect_str)
container_client = blob_service_client.get_container_client("Your Storage Name Here")

blob_list = container_client.list_blobs()
blob_client = blob_service_client.get_container_client(container= container_name)

# define worker function before a Pool is instantiated
def download_data(blob, local_path):
    download_file_path = os.path.join(local_path, blob.name.split('/')[2])
    with open(download_file_path, "wb") as download_file:
        download_file.write(blob_client.download_blob(blob.name).readall())

pool_size = 20 # our "parallelness"
pool = Pool(pool_size)
for blob in blob_list:
    pool.apply_async(download_data, (blob, local_path,))
    
pool.close()
pool.join()

Hope this was useful, feel free to share the best practices for similar scenarios. This way parallelization can be achieved for AWS S3 bucket and Google Cloud Storage as well.

Shadab Hussain0
Related