Введение

В этом примере показаны необходимые шаги для настройки пакетной операции над группой массовых спектров, содержащихся в одном или нескольких директории. Добиться этого можно последовательно или параллельно с помощью многоядерного компьютера или кластера компьютеров. Пакетная обработка адаптируется к однопрограммной модели параллельных вычислений с несколькими данными (SPMD), и она лучше всего подходит для Parallel Computing Toolbox™ и MATLAB ® Parallel Server™.

Сигналы для предварительной обработки поступают от усиленной поверхностью белка лазерной десорбции/времени ионизации массы рейса (SELDI-TOF) спектров. Данные в этом примере получены из Банка данных программы клинической протеомики FDA-NCI. В частности, в примере используется набор данных о раке яичников высокого разрешения, который был сгенерирован с использованием WCX2 массива белков. Подробное описание этого набора данных см. в разделах [1] и [2].

Настройка репозитория для данных

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

Сначала получите имя и полный путь к репозиторию данных. Заданы две строки: путь от локального компьютера до репозитория и путь, требуемый компьютерами кластера для доступа к одной и той же директории. Измените оба типа на сетевые строения.

local_repository  = 'C:/Examples/MassSpecRepository/OvarianCD_PostQAQC/';
worker_repository = 'L:/Examples/MassSpecRepository/OvarianCD_PostQAQC/';

В данном конкретном примере файлы хранятся в двух подкаталогах: 'Normal' и 'Cancer'. Можно создать списки, содержащие файлы для обработки с помощью DIR команда,

cancerFiles = dir([local_repository 'Cancer/*.txt'])
normalFiles = dir([local_repository 'Normal/*.txt'])

cancerFiles = 

  121×1 struct array with fields:

    name
    folder
    date
    bytes
    isdir
    datenum


normalFiles = 

  95×1 struct array with fields:

    name
    folder
    date
    bytes
    isdir
    datenum

и поместите их в одну переменную:

files = [ strcat('Cancer/',{cancerFiles.name}) ...
          strcat('Normal/',{normalFiles.name})];
N = numel(files)   % total number of files

N =

   216

Последовательная пакетная обработка

Прежде чем пытаться обработать все файлы параллельно, вам нужно протестировать свои алгоритмы локально с циклом for.

Напишите функцию с последовательным набором инструкций, которые нужно применить к каждому набору данных. Входные параметры - путь к данным (в зависимости от того, как их увидит машина, которая на самом деле сделает работу) и список файлов для обработки. Выходными аргументами являются предварительно обработанные сигналы и вектор M/Z. Поскольку вектор M/Z одинаков для каждой спектрограммы после предварительной обработки, его нужно хранить только один раз. Для примера:

type msbatchprocessing

function [MZ,Y] = msbatchprocessing(repository,files)
% MSBATCHPROCESSING Example function for BIODISTCOMPDEMO
%
% [MZ,Y] = MSBATCHPROCESSING(REPOSITORY,FILES) Preprocesses the
% spectrogram in files FILES and returns the mass/charge (MZ) and ion
% intensities (Y) vectors.
%
% Hard-coded parameters in the preprocessing steps have been adjusted to
% deal with the high-resolution spectrograms of the example.

% Copyright 2004-2012 The MathWorks, Inc.

K = numel(files);
Y = zeros(15000,K); % need to preset the size of Y for memory performance
MZ = zeros(15000,1);

parfor k = 1:K

    file = [repository files{k}]; 
    
    % read the two-column text file with mass-charge and intensity values
    fid = fopen(file,'r');
    ftext = textscan(fid, '%f%f');
    fclose(fid);
    signal = ftext{1};
    intensity = ftext{2};

    % resample the signal to 15000 points between 2000 and 11900
    mzout = (sqrt(2000)+(0:(15000-1))'*diff(sqrt([2000,11900]))/15000).^2;
    [mz,YR] = msresample(signal,intensity,mzout);
    
    % align the spectrograms to two good reference peaks
    P = [3883.766 7766.166];
    YA = msalign(mz,YR,P,'WIDTH',2);

    % estimate and adjust the background
    YB = msbackadj(mz,YA,'STEP',50,'WINDOW',50);

    % reduce the noise using a nonparametric filter
    Y(:,k) = mslowess(mz,YB,'SPAN',5);
    
    % the mass/charge vector is the same for all spectra after the resample
    if k==1
        MZ(:,k) = mz;
    end

end

Обратите внимание на функцию MSBATCHPROCESSING преднамеренное использование PARFOR вместо FOR. Пакетная обработка обычно реализуется задачами, которые являются независимыми между итерациями. В таком случае оператор FOR можно безразлично изменить на PARFOR, создание последовательности операторов MATLAB ® (или программы), которая может запускаться плавно на последовательном компьютере, многоядерном компьютере или кластере компьютеров без необходимости его изменения. В этом случае цикл выполняется последовательно, потому что вы не создали Параллельный пул (принимая, что в Parallel Computing Toolbox™ Preferences флажок для автоматического создания Параллельного пула не установлен, в противном случае MATLAB будет выполняться параллельно). Для примерных целей предварительно обработано и сохранено в Y только 20 спектрограмм матрица. Можно измерить количество времени, которое MATLAB ® занимает, чтобы завершить цикл с помощью TIC и TOC команды.

tic
repository = local_repository;
K = 20; % change to N to do all

[MZ,Y] = msbatchprocessing(repository,files(1:K));

disp(sprintf('Sequential time for %d spectrograms: %f seconds',K,toc))

Sequential time for 20 spectrograms: 7.725275 seconds

Параллельная пакетная обработка с многоядерными компьютерами

Если у вас есть Parallel Computing Toolbox™, можно использовать локальные рабочие места для параллелизации итераций цикла. Например, если ваша локальная машина имеет четыре ядра, можно запустить Parallel Pool с четырьмя рабочими, используя профиль кластера 'local' по умолчанию:

POOL = parpool('local',4);

tic
repository = local_repository;
K = 20; % change to N to do all

[MZ,Y] = msbatchprocessing(repository,files(1:K));

disp(sprintf('Parallel time with four local workers for %d spectrograms: %f seconds',K,toc))

Starting parallel pool (parpool) using the 'local' profile ...
Connected to the parallel pool (number of workers: 4).
Parallel time with four local workers for 20 spectrograms: 3.549382 seconds

Остановите локальный пул рабочих процессов:

delete(POOL)

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

Если у вас есть Parallel Computing Toolbox™ и MATLAB ® Parallel Server™ можно также распределить итерации цикла на большее количество компьютеров. В этом примере профиль кластера 'compbio _ config _ 01' соединяется с 6 работниками. Для получения информации о настройке и выборе параллельных строений смотрите «Профили кластеров и масштабирование расчетов» в документации Parallel Computing Toolbox™.

Обратите внимание, что если вы написали свою собственную функцию пакетной обработки, ее следует включить в соответствующий профиль кластера с помощью Диспетчера профилей кластеров. Это гарантирует, что MATLAB ® правильно передаст вашу новую функцию работникам. Доступ к Диспетчеру профилей кластеров осуществляется с помощью раскрывающегося меню Parallel на рабочем столе MATLAB ®.

POOL = parpool('compbio_config_01',6);

tic
repository = worker_repository;
K = 20; % change to N to do all

[MZ,Y] = msbatchprocessing(repository,files(1:K));

disp(sprintf('Parallel time with 6 remote workers for %d spectrograms: %f seconds',K,toc))

Starting parallel pool (parpool) using the 'compbio_config_01' profile ...
Connected to the parallel pool (number of workers: 6).
Parallel time with 6 remote workers for 20 spectrograms: 3.541660 seconds

Остановите пул кластеров:

delete(POOL)

Асинхронная параллельная пакетная обработка

Описанные выше схемы выполнения работают синхронно, то есть блокируют командную строку MATLAB ® до завершения их выполнения. Если вы хотите запустить задание пакетного процесса и получить доступ к командной строке, пока расчеты выполняются асинхронно (асинхронно), можно вручную распределить параллельные задачи и собрать результаты позже. Этот пример использует тот же профиль кластера, что и ранее.

Создайте одно задание с одной задачей (MSBATCHPROCESSING). Задача запускается на одном из рабочих и ее внутренних PARFOR цикл распределяется между всеми доступными рабочими в параллельном строении. Обратите внимание, что если N (количество спектрограмм) намного больше, чем количество доступных работников в вашем параллельном строении, Parallel Computing Toolbox™ автоматически балансирует рабочую нагрузку, даже если у вас гетерогенный кластер.

tic % start the clock
repository = worker_repository;
K = N; % do all spectrograms
CLUSTER = parcluster('compbio_config_01');
JOB = createCommunicatingJob(CLUSTER,'NumWorkersRange',[6 6]);
TASK = createTask(JOB,@msbatchprocessing,2,{repository,files(1:K)});

submit(JOB)

Когда задание отправлено, ваше местное приглашение MATLAB ® немедленно возвращается. Ваше параллельное задание запускается, как только параллельные ресурсы становятся доступными. Между тем, вы можете контролировать свое параллельное задание, проверяя TASK или JOB объекты. Используйте WAIT метод для программного ожидания, пока ваша задача не будет завершена:

wait(TASK)
TASK.OutputArguments

ans =

  1×2 cell array

    {15000×1 double}    {15000×216 double}

MZ = TASK.OutputArguments{1};
Y = TASK.OutputArguments{2};
destroy(JOB) % done retrieving the results
disp(sprintf('Parallel time (asynchronous) with 6 remote workers for %d spectrograms: %f seconds',K,toc))

Parallel time (asynchronous) with 6 remote workers for 216 spectrograms: 68.368132 seconds

Постобработка

После сбора всех данных можно использовать их локально. Для примера можно применить нормализацию группы:

Y = msnorm(MZ,Y,'QUANTILE',0.5,'LIMITS',[3500 11000],'MAX',50);

Создайте вектор группировки с типом для каждой спектрограммы, а также векторов индексации. Эта «маркировка» поможет выполнить последующий анализ на наборе данных.

grp = [repmat({'Cancer'},size(cancerFiles));...
       repmat({'Normal'},size(normalFiles))];
cancerIdx = find(strcmp(grp,'Cancer'));
numel(cancerIdx) % number of files in the "Cancer" subdirectory

ans =

   121

normalIdx = find(strcmp(grp,'Normal'));
numel(normalIdx) % number of files in the "Normal" subdirectory

ans =

    95

После маркировки данных можно отобразить некоторые спектрограммы каждого класса с использованием другого цвета (первые пять групп в этом примере).

h = plot(MZ,Y(:,cancerIdx(1:5)),'b',MZ,Y(:,normalIdx(1:5)),'r');
axis([7650 8200 -2 50])
xlabel('Mass/Charge (M/Z)');ylabel('Relative Intensity')
legend(h([1 end]),{'Ovarian Cancer','Control'})
title('Region of the pre-processed spectrograms')

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

save OvarianCancerQAQCdataset.mat Y MZ grp

Отказ от ответственности

TIC - TOC синхронизация представлена здесь в качестве примера. Последовательное и параллельное время выполнения будут варьироваться в зависимости от используемого оборудования.