Скопируйте несколько Azure контейнеров во вновь созданные контейнеры эффективно с помощью Python

Question

Я копирую содержимое из нескольких контейнеров в обозревателе хранилищ Azure и записываю это в кучу новых контейнеров и хочу узнать, как наиболее эффективно это сделать.

Существующие контейнеры называются cycling-input-1, cycling-input-2, .... и содержимое записывается в новые контейнеры, называемые cycling-output-1, cycling-output-2 и т. Д. Все контейнеры являются одного и того же типа (JPEG).

Приведенный ниже цикл for создает новый контейнер (циклический вывод) с требуемым суффиксом, а затем копирует большие двоичные объекты из соответствующего контейнера циклического ввода сюда. У меня есть около 30 контейнеров с тысячами изображений в каждом, так что я не уверен, что это лучший способ сделать это (это медленно). Есть ли лучший способ сделать это?

from azure.storage.blob.baseblobservice import BaseBlobService
account_name   = 'name'
account_key    = 'key'

# connect to the storage account
blob_service = BaseBlobService(account_name = account_name, account_key = account_key)

# get a list of the containers that need to be processed
cycling_containers = blob_service.list_containers(prefix = 'cycling-input')

# check the list of containers
for c in cycling_containers:
    print(c.name)


# copy across the blobs from existing containers to new containers with a prefix cycling-output 
prefix_of_new_container = 'cycling-output-'

for c in cycling_containers:
    contname = c.name
    generator = blob_service.list_blobs(contname)
    container_index = ''.join(filter(str.isdigit, contname))
    for blob in generator:
        flag_of_new_container = blob_service.create_container("%s%s" % (prefix_of_new_container, container_index))
        blob_service.copy_blob("%s%s" % (prefix_of_new_container, container_index), blob.name, "https://%s.blob.core.windows.net/%s/%s" % (account_name, contname, blob.name))

Answer 1

Простой способ - использовать модуль multiprocessing для параллельного копирования этих больших двоичных объектов всех контейнеров в их новые контейнеры, названные путем замены input на output.

. Вот мой пример кода в качестве ссылки.

from azure.storage.blob.baseblobservice import BaseBlobService
import multiprocessing

account_name = '<your account name>'
account_key = '<your account key>'

blob_service = BaseBlobService(
    account_name=account_name,
    account_key=account_key
)

cycling_containers = blob_service.list_containers(prefix = 'cycling-input')

def putBlobCopyTriples(queue, num_of_workers):
    for c in cycling_containers:
        container_name = c.name
        new_container_name = container_name.replace('input', 'output')
        blob_service.create_container(new_container_name)
        for blob in blob_service.list_blobs(container_name):
            blob_url = "https://%s.blob.core.windows.net/%s/%s" % (account_name, container_name, blob.name)
            queue.put( (new_container_name, blob.name, blob_url) )
    for i in range(num_of_workers):
        queue.put( (None, None, None) )

def copyWorker(lock, queue, sn):
    while True:
        with lock:
            (new_container_name, blob_name, new_blob_url) = queue.get()
        if new_container_name == None:
            break
        print(sn, new_container_name, blob_name, new_blob_url)
        blob_service.copy_blob(new_container_name, blob_name, new_blob_url)

if __name__ == '__main__':
    num_of_workers = 4 # the number of workers what you want, for example, 4 is my cpu core count
    lock = multiprocessing.Lock()
    queue = multiprocessing.Queue()
    multiprocessing.Process(target = putBlobCopyTriples, args = (queue, num_of_workers)).start()
    workers = [multiprocessing.Process(target = copyWorker, args = (lock, queue, i)) for i in range(num_of_workers)]
    for p in workers:
        p.start()

Примечание. Кроме числа ядер процессора в вашей среде, ограничения скорости копирования зависят от пропускной способности ввода-вывода.Число рабочих не больше, тем лучше.Рекомендуется, чтобы это число было равно или меньше вашего числа процессоров или числа гиперпоточности.