Сеть поездов параллельно с индивидуальным контуром обучения

В этом примере используются:

В этом примере показано, как настроить пользовательский цикл обучения для параллельного обучения сети. В этом примере параллельные рабочие тренируются на участках всей мини-партии. Если у вас есть ГПУ, то обучение происходит на ГПУ. Во время обучения, DataQueue объект отправляет информацию о ходе обучения обратно клиенту MATLAB.

Загрузить набор данных

Загрузите набор данных цифр и создайте хранилище данных изображения для набора данных. Разделите хранилище данных на обучающие и тестовые хранилища данных рандомизированным способом.

digitDatasetPath = fullfile(matlabroot,'toolbox','nnet','nndemos', ...
    'nndatasets','DigitDataset');
imds = imageDatastore(digitDatasetPath, ...
    'IncludeSubfolders',true, ...
    'LabelSource','foldernames');

[imdsTrain,imdsTest] = splitEachLabel(imds,0.9,"randomized");

Определите различные классы в обучающем наборе.

classes = categories(imdsTrain.Labels);
numClasses = numel(classes);

Определение сети

Определите сетевую архитектуру и создайте график слоев с помощью layerGraph функция. Эта сетевая архитектура включает уровни пакетной нормализации, которые отслеживают среднее значение и статистику дисперсии набора данных. При параллельном обучении объедините статистику всех работников в конце каждого шага итерации, чтобы состояние сети отражало весь мини-пакет. В противном случае состояние сети может расходиться по рабочим. Если вы обучаете рекуррентные нейронные сети (RNN), например, используя данные последовательности, которые были разбиты на меньшие последовательности, для обучения сетей, содержащих уровни LSTM или GRU, вы также должны управлять состоянием между рабочими.

layers = [
    imageInputLayer([28 28 1],'Name','input','Normalization','none')
    convolution2dLayer(5,20,'Name','conv1')
    batchNormalizationLayer('Name','bn1')
    reluLayer('Name','relu1')
    convolution2dLayer(3,20,'Padding',1,'Name','conv2')
    batchNormalizationLayer('Name','bn2')
    reluLayer('Name','relu2')
    convolution2dLayer(3,20,'Padding',1,'Name','conv3')
    batchNormalizationLayer('Name','bn3')
    reluLayer('Name','relu3')
    fullyConnectedLayer(numClasses,'Name','fc')];

lgraph = layerGraph(layers);

Создать dlnetwork объект из графа слоев. dlnetwork объекты позволяют проводить обучение с использованием пользовательских циклов.

dlnet = dlnetwork(lgraph)

dlnet = 
  dlnetwork with properties:

         Layers: [11×1 nnet.cnn.layer.Layer]
    Connections: [10×2 table]
     Learnables: [14×3 table]
          State: [6×3 table]
     InputNames: {'input'}
    OutputNames: {'fc'}
    Initialized: 1

Настройка параллельной среды

Определите, доступны ли графические процессоры для MATLAB для использования с canUseGPU функция.

Если имеются GPU, то выполните обучение на GPU. Создайте параллельный пул с таким количеством работников, как GPU.
Если нет доступных графических процессоров, выполните обучение на процессорах. Создайте параллельный пул с числом работников по умолчанию.

if canUseGPU
    executionEnvironment = "gpu";
    numberOfGPUs = gpuDeviceCount("available");
    pool = parpool(numberOfGPUs);
else
    executionEnvironment = "cpu";
    pool = parpool;
end

Получите количество работников в параллельном пуле. Позднее в этом примере рабочая нагрузка будет разделена по этому номеру.

N = pool.NumWorkers;

Модель поезда

Укажите параметры обучения.

numEpochs = 20;
miniBatchSize = 128;
velocity = [];

Для обучения GPU рекомендуется увеличивать размер мини-партии линейно с количеством GPU, чтобы поддерживать постоянную рабочую нагрузку на каждом GPU. Дополнительные сведения см. в разделе Обучение с использованием нескольких графических процессоров.

if executionEnvironment == "gpu"
    miniBatchSize = miniBatchSize .* N
end

miniBatchSize = 512

Рассчитайте размер мини-партии для каждого работника путем равномерного деления общего размера мини-партии между работниками. Распределите остаток по первым работникам.

workerMiniBatchSize = floor(miniBatchSize ./ repmat(N,1,N));
remainder = miniBatchSize - sum(workerMiniBatchSize);
workerMiniBatchSize = workerMiniBatchSize + [ones(1,remainder) zeros(1,N-remainder)]

workerMiniBatchSize = 1×4

   128   128   128   128

Инициализируйте график хода обучения.

% Set up the training plot
figure
lineLossTrain = animatedline('Color',[0.85 0.325 0.098]);
ylim([0 inf])
xlabel("Iteration")
ylabel("Loss")
grid on

Чтобы отправить данные обратно от работников во время обучения, создайте DataQueue объект. Использовать afterEach для настройки функции, displayTrainingProgress, чтобы вызывать каждый раз, когда работник отправляет данные. displayTrainingProgress - вспомогательная функция, определенная в конце этого примера, которая отображает информацию о ходе обучения, поступающую от работников.

Q = parallel.pool.DataQueue;
displayFcn = @(x) displayTrainingProgress(x,lineLossTrain);
afterEach(Q,displayFcn);

Выполните обучение модели с помощью пользовательского параллельного цикла обучения, как описано в следующих шагах. Чтобы выполнить код одновременно для всех работников, используйте spmd блок. В пределах spmd блок, labindex возвращает индекс работника, выполняющего в данный момент код.

Перед обучением разделите хранилище данных для каждого работника с помощью partition функция и установка ReadSize к размеру мини-партии работника.

Для каждой эпохи сбросьте и перетасуйте хранилище данных с помощью reset и shuffle функции. Для каждой итерации в эпохе:

Убедитесь, что все работники имеют доступные данные до начала параллельной обработки, выполнив глобальный and эксплуатация (gop) о результате hasdata функция.
Считывание мини-пакета из хранилища данных с помощью read и объединяют извлеченные изображения в четырехмерный массив изображений. Нормализуйте изображения таким образом, чтобы пикселы принимали значения между 0 и 1.
Преобразуйте метки в матрицу фиктивных переменных, которая сопоставляет метки с наблюдениями. Фиктивные переменные содержат 1 для метки наблюдения и 0 в противном случае.
Преобразование мини-пакета данных в dlarray объект с базовым типом один и укажите метки размеров 'SSCB' (пространственный, пространственный, канальный, пакетный). Для обучения GPU преобразуйте данные в gpuArray.
Вычислить градиенты и потери сети на каждом работнике путем вызова dlfeval на modelGradients функция. dlfeval функция оценивает вспомогательную функцию modelGradients с включенным автоматическим различением, так modelGradients может вычислять градиенты относительно потерь автоматически. modelGradients определяется в конце примера и возвращает потери и градиенты с учетом сети, мини-пакета данных и истинных меток.
Для получения общей потери агрегируйте потери по всем работникам. В этом примере для функции потерь используется перекрестная энтропия, а агрегированные потери представляют собой сумму всех потерь. Перед агрегированием нормализуйте каждую потерю путем умножения на долю общей мини-партии, над которой работает работник. Использовать gplus для объединения всех потерь и тиражирования результатов среди работников.
Для агрегирования и обновления градиентов всех работников используйте dlupdate функции с помощью aggregateGradients функция. aggregateGradients является вспомогательной функцией, определенной в конце этого примера. Эта функция использует gplus для объединения и тиражирования градиентов между работниками после нормализации в соответствии с долей общего мини-пакета, над которым работает каждый работник.
Агрегировать состояние сети для всех работников, использующих aggregateState функция. aggregateState является вспомогательной функцией, определенной в конце этого примера. Уровни пакетной нормализации в сети отслеживают среднее значение и дисперсию данных. Поскольку полная мини-партия распределяется по нескольким работникам, агрегируйте состояние сети после каждой итерации, чтобы вычислить среднее значение и дисперсию всей минибатчи.
После вычисления конечных градиентов обновите сетевые обучаемые параметры с помощью sgdmupdate функция.
Отправка информации о ходе обучения клиенту с помощью send функции с помощью DataQueue. Используйте только одного работника для отправки данных, поскольку все работники имеют одинаковую информацию о потерях. Чтобы убедиться, что данные находятся на CPU, чтобы клиентская машина без графического процессора могла получить к ним доступ, используйте gather на dlarray перед отправкой.

start = tic;
spmd
    % Partition datastore.
    workerImds = partition(imdsTrain,N,labindex);
    workerImds.ReadSize = workerMiniBatchSize(labindex);
    
    workerVelocity = velocity;
   
    iteration = 0;
    
    for epoch = 1:numEpochs
        % Reset and shuffle the datastore.
        reset(workerImds);
        workerImds = shuffle(workerImds);
        
        % Loop over mini-batches.
        while gop(@and,hasdata(workerImds))
            iteration = iteration + 1;
            
            % Read a mini-batch of data.
            [workerXBatch,workerTBatch] = read(workerImds);
            workerXBatch = cat(4,workerXBatch{:});
            workerNumObservations = numel(workerTBatch.Label);

            % Normalize the images.
            workerXBatch =  single(workerXBatch) ./ 255;
            
            % Convert the labels to dummy variables.
            workerY = zeros(numClasses,workerNumObservations,'single');
            for c = 1:numClasses
                workerY(c,workerTBatch.Label==classes(c)) = 1;
            end
            
            % Convert the mini-batch of data to dlarray.
            dlworkerX = dlarray(workerXBatch,'SSCB');
            
            % If training on GPU, then convert data to gpuArray.
            if executionEnvironment == "gpu"
                dlworkerX = gpuArray(dlworkerX);
            end
            
            % Evaluate the model gradients and loss on the worker.
            [workerGradients,dlworkerLoss,workerState] = dlfeval(@modelGradients,dlnet,dlworkerX,workerY);
            
            % Aggregate the losses on all workers.
            workerNormalizationFactor = workerMiniBatchSize(labindex)./miniBatchSize;
            loss = gplus(workerNormalizationFactor*extractdata(dlworkerLoss));
            
            % Aggregate the network state on all workers
            dlnet.State = aggregateState(workerState,workerNormalizationFactor);
            
            % Aggregate the gradients on all workers.
            workerGradients.Value = dlupdate(@aggregateGradients,workerGradients.Value,{workerNormalizationFactor});
            
            % Update the network parameters using the SGDM optimizer.
            [dlnet.Learnables,workerVelocity] = sgdmupdate(dlnet.Learnables,workerGradients,workerVelocity);
        end
        
       % Display training progress information.
       if labindex == 1
           data = [epoch loss iteration toc(start)];
           send(Q,gather(data)); 
       end
    end
end

Тестовая модель

После обучения сети можно проверить ее точность.

Загрузите тестовые образы в память с помощью readall в хранилище тестовых данных, объедините их и нормализуйте.

XTest = readall(imdsTest);
XTest = cat(4,XTest{:});
XTest = single(XTest) ./ 255;
YTest = imdsTest.Labels;

После завершения обучения все работники имеют одинаковую полную обученную сеть. Извлеките любой из них.

dlnetFinal = dlnet{1};

Классификация изображений с помощью dlnetwork объект, используйте predict функция на dlarray.

dlYPredScores = predict(dlnetFinal,dlarray(XTest,'SSCB'));

Из прогнозируемых баллов найдите класс с наивысшим баллом с max функция. Перед этим извлеките данные из dlarray с extractdata функция.

[~,idx] = max(extractdata(dlYPredScores),[],1);
YPred = classes(idx);

Чтобы получить точность классификации модели, сравните прогнозы на тестовом наборе с истинными метками.

accuracy = mean(YPred==YTest)

accuracy = 0.9910

Функции градиентов модели

Определите функцию, modelGradientsдля вычисления градиентов потерь относительно обучаемых параметров сети. Эта функция вычисляет сетевые выходы для мини-пакета X с forward и softmax и вычисляет потери, учитывая истинные выходы, используя перекрестную энтропию. При вызове этой функции с dlfeval, автоматическое дифференцирование включено, и dlgradient может вычислять градиенты потерь по отношению к обучаемым.

function [dlgradients,dlloss,state] = modelGradients(dlnet,dlX,dlY)
    [dlYPred,state] = forward(dlnet,dlX);
    dlYPred = softmax(dlYPred);
    
    dlloss = crossentropy(dlYPred,dlY);
    dlgradients = dlgradient(dlloss,dlnet.Learnables);
end

Просмотр функции «Ход обучения»

Определите функцию для отображения информации о ходе обучения, поступающей от работников. DataQueue в этом примере эта функция вызывается каждый раз, когда работник отправляет данные.

function displayTrainingProgress (data,line)
     addpoints(line,double(data(3)),double(data(2)))
     D = duration(0,0,data(4),'Format','hh:mm:ss');
     title("Epoch: " + data(1) + ", Elapsed: " + string(D))
     drawnow
end

Функция агрегирования градиентов

Определите функцию, которая агрегирует градиенты на всех рабочих, добавив их вместе. gplus добавляет и реплицирует все градиенты на рабочих. Прежде чем сложить их вместе, нормализуйте, умножив их на коэффициент, который представляет долю общего мини-пакета, над которым работает работник. Извлечение содержимого dlarray, use extractdata.

function gradients = aggregateGradients(dlgradients,factor)
    gradients = extractdata(dlgradients);
    gradients = gplus(factor*gradients);
end

Агрегатная функция состояния

Определите функцию, которая агрегирует состояние сети для всех работников. Состояние сети содержит обученную статистику нормализации пакета набора данных. Поскольку каждый работник видит только часть мини-пакета, агрегируйте состояние сети таким образом, чтобы статистика представляла статистику по всем данным. Для каждой мини-партии комбинированное среднее рассчитывается как средневзвешенное среднее для работников для каждой итерации. Комбинированную дисперсию рассчитывают по следующей формуле:

$_{}^{} \frac{}{}_{}^{}_{}_{}^{} sc2=1M\sumj=1Nmj[sj2+ {{\overset{}{(}}_{} {\overset{}{}}_{} x‾j-x‾c}^{)}$ 2]

где N- общее число работников, M- общее число наблюдений в мини-партии, $_{mj}$ - количество наблюдений, обработанных на j-ом работнике, ${\overset{}{}}_{x‾j}$ и $_{}^{sj2}$ - статистика среднего и дисперсии, рассчитанная на этом работнике, и ${\overset{}{}}_{x‾c}$ - объединенное среднее по всем работникам.

function state = aggregateState(state,factor)

    numrows = size(state,1);
    
    for j = 1:numrows
        isBatchNormalizationState = state.Parameter(j) =="TrainedMean"...
            && state.Parameter(j+1) =="TrainedVariance"...
            && state.Layer(j) == state.Layer(j+1);
        
        if isBatchNormalizationState
            meanVal = state.Value{j};
            varVal = state.Value{j+1};
            
            % Calculate combined mean
            combinedMean = gplus(factor*meanVal);
                   
            % Caclulate combined variance terms to sum
            combinedVarTerm = factor.*(varVal + (meanVal - combinedMean).^2);        
            
            % Update state
            state.Value(j) = {combinedMean};
            state.Value(j+1) = {gplus(combinedVarTerm)};
           
        end
    end
end

См. также

Документация