Пример при развертывании длинных массивов к Spark Enabled кластер Hadoop

Поддерживаемая платформа: Linux^® только.

Этот пример показывает, как развернуться, приложение MATLAB^®, содержащее длинные массивы к Spark™, включило кластер Hadoop^®.

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

Набор данных:	`airlinesmall.csv`
Описание:	Отъезд авиакомпании и информация о прибытии от 1987-2008.
Местоположение :	`/usr/local/MATLAB/R2019a/toolbox/matlab/demos`

Примечание

Можно следовать тем же инструкциям, чтобы развернуть приложения Spark длинного массива на Cloudera^® CDH. Чтобы видеть пример на MATLAB Answers™, щелкнуть здесь.

Чтобы использовать зоны шифрования Cloudera CDH, добавьте файл JAR commons-codec-1.9.jar в статический путь к классу MATLAB Runtime. Местоположение файла: $HADOOP_PREFIX/lib/commons-codec-1.9.jar, где $HADOOP_PREFIX является местоположением, где Hadoop установлен.

Примечание

Если вы будете использовать версию 1.6 Spark или выше, необходимо будет увеличить размер кучи Java^® в MATLAB по крайней мере к 512 МБ. Для получения информации о том, как увеличить размер кучи Java в MATLAB, смотрите Настройки Java Heap Memory (MATLAB).

Предпосылки

Запустите этот пример путем создания новой папки работы, которая видима к пути поиска файлов MATLAB.
Установите MATLAB Runtime в папке, которая доступна каждым узлом рабочего в кластере Hadoop. Этот пример использует /usr/local/MATLAB/MATLAB_Runtime/v## в качестве местоположения папки MATLAB Runtime.
Если у вас нет MATLAB Runtime, можно загрузить его с веб-сайта в: https://www.mathworks.com/products/compiler/mcr.
Примечание
Замените все ссылки на версию MATLAB Runtime v## в этом примере с номером версии MATLAB Runtime, соответствующим вашему релизу MATLAB. Например, MATLAB R2017b имеет номер версии MATLAB Runtime v92. Для получения информации о номерах версий MATLAB Runtime соответствующие релизы MATLAB см. этот список.
Скопируйте файл airlinesmall.csv в папку Hadoop Distributed File System (HDFS™) /user/<username>/datasets. Здесь <username> относится к вашему имени пользователя в HDFS.
$ ./hadoop fs -copyFromLocal airlinesmall.csv hdfs://host:54310/user/<username>/datasets

Процедура

Настройте переменную окружения, HADOOP_PREFIX, чтобы указать на вашу папку установки Hadoop. Эти свойства необходимы для представления заданий к вашему кластеру Hadoop.
```
setenv('HADOOP_PREFIX','/usr/lib/hadoop')
```
Переменная окружения HADOOP_PREFIX должна собираться при использовании функции datastore MATLAB указать на данные по HDFS. Установка этой переменной окружения не имеет никакого отношения к Spark. Смотрите Отношение Между Spark и Hadoop для получения дополнительной информации.
Если вы планируете использование набора данных, это находится на вашей локальной машине в противоположность одной на HDFS, то можно пропустить этот шаг.
Примечание
Этот пример использует /usr/lib/hadoop в качестве директории, где Hadoop установлен. Ваша директория установки Hadoop, возможно, отличающаяся.
Свойства Specify Spark.
Используйте объект containers.Map задать свойства Spark.
```
sparkProperties = containers.Map( ...
 {'spark.executor.cores', ...
 'spark.executor.memory', ...
 'spark.yarn.executor.memoryOverhead', ...
 'spark.dynamicAllocation.enabled', ...
 'spark.shuffle.service.enabled', ...
 'spark.eventLog.enabled', ...
 'spark.eventLog.dir'}, ...
 {'1', ...
  '2g', ...
  '1024', ...
  'true', ...
  'true', ...
  'true', ...
  'hdfs://host:54310/user/<username>/log'});
```
Для получения дополнительной информации о свойствах Spark расширьте значение prop пары "имя-значение" 'SparkProperties' в разделе Input Arguments класса SparkConf. Класс SparkConf является частью MATLAB API для Spark, который обеспечивает альтернативный способ развернуть приложения MATLAB на Spark. Для получения дополнительной информации смотрите, Развертывают Приложения Используя MATLAB API для Spark.
Сконфигурируйте свое приложение MATLAB, содержащее длинные массивы с параметрами Spark.
Используйте класс matlab.mapreduce.DeploySparkMapReducer, чтобы сконфигурировать ваше приложение MATLAB, содержащее длинные массивы с параметрами Spark как пары "ключ-значение".
```
conf = matlab.mapreduce.DeploySparkMapReducer( ...
      'AppName','myTallApp', ...
      'Master','yarn-client', ...
      'SparkProperties',sparkProperties);
```
Для получения дополнительной информации смотрите matlab.mapreduce.DeploySparkMapReducer.
Задайте среду выполнения Spark.
Используйте функцию mapreducer, чтобы задать среду выполнения Spark.
```
mapreducer(conf)
```
Для получения дополнительной информации смотрите mapreducer.

Включайте свой код приложения MATLAB, содержащий длинные массивы.

Используйте функцию MATLAB datastore, чтобы создать объект datastore, указывающий на файл airlinesmall.csv в HDFS. Передайте объект datastore как входной параметр к функции tall. Это создаст длинный массив. Можно выполнить операции на длинном массиве, чтобы вычислить среднюю задержку прибытия и самые большие задержки прибытия.

% Create a |datastore| for a collection of tabular text files representing airline data. 
% Select the variables of interest, specify a categorical data type for the 
% |Origin| and |Dest| variables.
% ds = datastore('airlinesmall.csv') % if using a dataset on your local machine
ds = datastore('hdfs:///<username>/datasets/airlinesmall.csv');
ds.TreatAsMissing = 'NA';
ds.SelectedVariableNames = {'Year','Month','ArrDelay','DepDelay','Origin','Dest'};
ds.SelectedFormats(5:6) = {'%C','%C'};

% Create Tall Array
% Tall arrays are like normal MATLAB arrays, except that they can have any
% number of rows. When a |tall| array is backed by a |datastore|, the underlying class of
% the tall array is based on the type of datastore.
tt = tall(ds);

% Remove Rows with Missing Data or NaN Values
idx = any(ismissing(tt),2); 
tt(idx,:) = [];

% Compute Mean Delay
meanArrivalDelay = mean(tt.DepDelay,'omitnan');
biggestDelays = topkrows(tt,10,'ArrDelay');

% Gather Results
% The |gather| function forces evaluation of all queued operations and
% brings the resulting output back into memory. 
[meanArrivalDelay,biggestDelays] = gather(meanArrivalDelay,biggestDelays)

% Delete mapreducer object
delete(conf);

Создайте приложение Spark.

Используйте команду mcc с опциями -vCW, чтобы создать приложение Spark.

>> mcc -vCW 'Spark:myTallApp' deployTallArrayToSpark.m

Следующие файлы создаются.

Файлы	Описание
`run_myTallApp.sh`	Сценарий оболочки, чтобы запустить приложение. Скрипт вызывает `spark-submit`, чтобы запустить приложение на кластере.
`myTallApp.jar`	JAR приложения. JAR приложения содержит, группировал код MATLAB и другие зависимости.
`readme.txt`	Файл Readme, содержащий детали о том, как запустить приложение.
`requiredMCRProducts.txt`
`mccExcludedFiles.log`

Для получения дополнительной информации смотрите mcc.

Запустите приложение от интерпретатора Linux с помощью следующей команды:
$ ./run_myTallApp.sh /usr/local/MATLAB/MATLAB_Runtime/v##
/usr/local/MATLAB/MATLAB_Runtime/v## является аргументом, указывающим на местоположение MATLAB Runtime.

Вы будете видеть следующий вывод:

meanArrivalDelay =
    7.1201

biggestDelays =

  10x5 table

    Year    Month    ArrDelay    Origin    Dest
    ____    _____    ________    ______    ____

    1995    11       1014        HNL       LAX 
    2007     4        914        JFK       DTW 
    2001     4        887        MCO       DTW 
    2008     7        845        CMH       ORD 
    1988     3        772        ORD       LEX 
    2008     4        710        EWR       RDU 
    1998    10        679        MCI       DFW 
    2006     6        603        ABQ       PHX 
    2008     6        586        PIT       LGA 
    2007     4        568        RNO       SLC

Опционально, если вы хотите анализировать или просмотреть результаты, сгенерированные вашим приложением в MATLAB, необходимо записать результаты в файл на HDFS использование функции write для длинных массивов. Можно затем считать файл с помощью функции datastore.

Чтобы записать результаты зарегистрировать на HDFS, добавьте следующую строку кода в свое приложение MATLAB незадолго до оператора delete(conf) и затем группируйте свое приложение:

write('hdfs:///user/<username>/results', tall(biggestDelays));

Замените <username> на свое имя пользователя.

Можно только сохранить одну переменную в файл с помощью функции write для длинных массивов. Поэтому необходимо будет записать в несколько файлов, если вы захотите сохранить несколько переменных.

Просмотреть результаты в MATLAB после выполнения приложения против Spark включило кластер, используйте функцию datastore можно следующим образом:

>> ds = datastore('hdfs:///user/<username>/results')
>> readall(ds)

Вы, возможно, должны установить переменную окружения HADOOP_PREFIX с помощью функционального setenv в случае, если вы не можете просмотреть результаты с помощью функции datastore.

Примечание

Если развертываемым приложением длинного массива является функция MATLAB в противоположность скрипту MATLAB, используйте следующий синтаксис выполнения:

$ ./run_<applicationName>.sh \
  <MATLAB_Runtime_Location> \
  [Spark arguments] \
  [Application arguments]

Например:

$ ./run_myTallApp.sh \
   /usr/local/MATLAB/MATLAB_Runtime/v92 \
   yarn-client \
   hdfs://host:54310/user/<username>/datasets/airlinesmall.csv \
   hdfs://host:54310/user/<username>/result

Код:

deployTallArrayToSpark.m

%% Set Hadoop environemnt variables
setenv('HADOOP_PREFIX','/usr/lib/hadoop')

%% Setup Spark Properties
% Use a containers.Map object to specify Spark properties.
sparkProperties = containers.Map( ...
 {'spark.executor.cores', ...
 'spark.executor.memory', ...
 'spark.yarn.executor.memoryOverhead', ...
 'spark.dynamicAllocation.enabled', ...
 'spark.shuffle.service.enabled', ...
 'spark.eventLog.enabled', ...
 'spark.eventLog.dir'}, ...
 {'1', ...
  '2g', ...
  '1024', ...
  'true', ...
  'true', ...
  'true', ...
  'hdfs://host:54310/user/<username>/log'});

%% Configure a MATLAB Tall Array Application
% Use the class matlab.mapreduce.DeploySparkMapReducer to configure the 
% MATLAB tall array application with Spark parameters as key-value pairs.
conf = matlab.mapreduce.DeploySparkMapReducer( ...
      'AppName','myTallApp', ...
      'Master','yarn-client', ...
      'SparkProperties',sparkProperties);

%% Define the Spark Execution Environment
% Use the mapreducer function to define the Spark execution environment.
mapreducer(conf);

%% Create a |datastore| for a collection of tabular text files representing airline data. 
% Select the variables of interest, specify a categorical data type for the 
% |Origin| and |Dest| variables.
ds = datastore('hdfs://host:54310/user/<username>/datasets/airlinesmall.csv');
ds.TreatAsMissing = 'NA';
ds.SelectedVariableNames = {'Year','Month','ArrDelay','Origin','Dest'};
ds.SelectedFormats(4:5) = {'%C','%C'};

%% Create Tall Array
% Tall arrays are like normal MATLAB arrays, except that they can have any
% number of rows. When a |tall| array is backed by a |datastore|, the underlying class of
% the tall array is based on the type of datastore.
tt = tall(ds);

%% Remove Rows with Missing Data or NaN Values
idx = any(ismissing(tt),2); 
tt(idx,:) = [];

%% Compute Mean Delay
meanArrivalDelay = mean(tt.ArrDelay,'omitnan');
biggestDelays = topkrows(tt,10,'ArrDelay');

%% Gather Results
% The |gather| function forces evaluation of all queued operations and
% brings the resulting output back into memory. 
[meanArrivalDelay,biggestDelays] = gather(meanArrivalDelay,biggestDelays)

%% Write results to HDFS
write('hdfs:///user/<username>/results', tall(meanArrivalDelay));
write('hdfs:///user/<username>/results', tall(biggestDelays));

%% Delete mapreducer object
delete(conf);

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