Предварительная загрузка части файла HDF5 при выполнении другой задачи

Question

Я тренирую глубокий классификатор обучения, который использует набор данных HDF5, который слишком велик, чтобы вписаться в память. Поэтому я извлекаю данные партиями 256 и использую эти партии для обучения моего классификатора следующим образом. Библиотека глубокого обучения, которую я использую (Keras), предоставляет метод model.train_on_batch(X_batch, y_batch).

for i in range(n_batches_in_dset): 
     X_batch, y_batch = load_partition('train', ind=[i*batch_size, (i+1)*batch_size]) 
     loss = model.train_on_batch(X_batch, y_batch) 

Было бы целесообразно предварительно отбирать следующую партию данных во время обучения текущим данным с использованием графического процессора. Как это можно сделать в Python?

Я приложил код, который я использую для загрузки данных.

def load_hdf5(path, datapart, ind=None): 
    f = h5py.File(path, 'r') 
    if ind is None: 
     dat = f[datapart][:] 
    else: 
     dat = f[datapart][ind[0]:ind[1]] 
    f.close() 
    return np.array(dat) 

def load_partition(name, ind=None): 
    path = DEEP_ROOT + 'data/{}.h5'.format(name) 
    X = load_hdf5(path, 'data', ind) 
    y = load_hdf5(path, 'label', ind) 
    X = np.swapaxes(X, 2, 3) 
    y = np_utils.to_categorical(y) 
    return X, y 

Answer 1

наверное проще всего сделать, это поставить отдельные задачи в отдельном threads, с синхронизированным queue передать партии между ними. Мы будем использовать отдельный поток для части считывания данных и основной поток для части обучения.

import Queue, threading 

data_queue = Queue.Queue(2) # a queue with two space for two "chunks" 
sentinel = object() 

#start the data-loading task 
def load_task() 
    for x in i in range(n_batches_in_dset): 
     data_queue.put(load_partition('train', ind=[i*batch_size, (i+1)*batch_size]), True) 
    # tell the other side we're "done" 
    data_queue.put(sentinel, True) 

threading.Thread(target=load_task).start() 

while True: 
    batch = data_queue.get(True) 
    data_queue.task_done() 
    if batch is sentinel: 
     break # we're done now! 
    X_batch, y_batch = batch 
    loss = model.train_on_batch(X_batch, y_batch) 

EDIT: мы должны использовать Queue.task_done() для разблокирования очередь нужно