I am writing a code to obtain the size of all Azure DataLake Storage files. I'm using a multithread process to reduce the execution time of the code.
The code does the following:
- It get the first 3 levels folders of my datalake:
- raw/project1/2020/01
- raw/project1/2020/02
- ...
- raw/projectN/2020/01
- It creates 4 threads
- It put each folder in a Queue
- For each folder I obtain all the files and the sizes in the folder
- It writes each folder result in a file
- It upload the file to a StorageAccount
def downloadFolder(adls_path):
#return
try:
return getAllSubdirectories(adls_client, adls_path)
except PermissionError:
#write folder into file
class DownloadWorker(Thread):
def __init__(self, queue):
Thread.__init__(self)
self.queue = queue
def run(self):
while True:
directory = self.queue.get()
try:
data = downloadFolder(directory)
if data is not None:
upload_file_to_blob(sa_tenant, sa_client, sa_secret,
sa_name, sa_container,
file_path)
else:
print(Empty folder)
finally:
self.queue.task_done()
if __name__ == "__main__":
adls_client = connectADLS(
sa_tenant(), sa_client, sa_secret, adls_name
rootFolders = getSubdirectories(adls_client, adls_root_dir)
random.shuffle(rootFolders)
queue = Queue()
for i in range(4):
worker = DownloadWorker(queue)
worker.daemon = True
worker.start()
for Folder in rootFolders:
queue.put((Folder))
queue.join()
sys.exit(0)
Monitoring the memory with memory_profiler I see an increase of the memory that never falls:
Why I am using 2000 MiB of memory if the biggest file created is over 300 MB and how can I control this memory usage?
