Пример на развертывающихся приложениях к Spark Используя MATLAB API для Spark

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

Этот пример показывает вам, как развернуть автономное приложение в Spark™ с помощью MATLAB^® API для Spark. Ваше приложение может быть развернуто против Spark с помощью одного из двух поддерживаемых кластерных менеджеров: локальный и Hadoop^® YARN. Этот пример показывает вам, как развернуть ваше приложение с помощью обоих кластерных менеджеров. Для обсуждения кластерных менеджеров смотрите Кластерных менеджеров, Поддержанных Spark.

Локальный
Hadoop YARN

Цель: Считайте количество уникальных авиакомпаний в данном наборе данных.

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

Функция помощника

Создайте файл MATLAB под названием carrierToCount.m со следующим кодом:

function results = carrierToCount(input)
    tbl = input{1};
    intermKeys = tbl.UniqueCarrier;
    [intermKeys, ~, idx] = unique(intermKeys);
    intermValues = num2cell(accumarray(idx, ones(size(idx))));
    results = cellfun( @(x,y) {x,y} , ...
        intermKeys, intermValues, ...
        'UniformOutput',false);

Эта функция помощника передается в как указатель на функцию к одному из методов в примере.

Примечание

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

Локальный

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

Необходимые условия

Запустите этот пример путем создания новой папки работы, которая отображается к пути поиска файлов MATLAB.
Создайте функцию помощника carrierToCount.m упомянутый выше.

Процедура

Свойства Specify Spark.
Используйте containers.Map объект задать свойства Spark.
```
sparkProp = containers.Map(...
    {'spark.executor.cores',...
    'spark.matlab.worker.debug'},...
    {'1',...
    'true'});
```
Свойства Spark указывают на среду выполнения Spark приложения, которое развертывается. Каждое приложение должно быть сконфигурировано с определенными свойствами Spark для него, чтобы быть развернутым.
Для получения дополнительной информации о свойствах Spark расширьте prop значение 'SparkProperties' пара "имя-значение" в разделе Input Arguments SparkConf класс.
Создайте SparkConf объект.
Используйте класс matlab.compiler.mlspark.SparkConf создать SparkConf объект. SparkConf объектно-ориентированная память параметры конфигурации приложения, развертываемого на Spark. Параметры конфигурации приложения передаются на кластер Spark через SparkContext.
```
conf = matlab.compiler.mlspark.SparkConf(...
    'AppName', 'mySparkAppDepLocal', ...
    'Master', 'local[1]', ...
    'SparkProperties', sparkProp );
```
Для получения дополнительной информации о SparkConf смотрите matlab.compiler.mlspark.SparkConf.
Создайте SparkContext объект.
Используйте класс matlab.compiler.mlspark.SparkContext с SparkConf возразите как вход, чтобы создать SparkContext объект.
```
sc = matlab.compiler.mlspark.SparkContext(conf);
```
SparkContext объект служит точкой входа к Spark путем инициализации связи с кластером Spark. Это принимает SparkConf возразите как входной параметр и использование параметры, заданные в том объекте настроить внутренние сервисы, необходимые, чтобы установить связь со средой выполнения Spark.
Для получения дополнительной информации о SparkContext смотрите matlab.compiler.mlspark.SparkContext.
Создайте RDD объект из данных.
Используйте функцию MATLAB datastore создать datastore объект, указывающий на файл airlinesmall.csv. Затем используйте метод SparkContext datastoreToRDD преобразовывать datastore возразите против Spark RDD объект.
```
% Create a MATLAB datastore (LOCAL)
ds = datastore('airlinesmall.csv',...
    'TreatAsMissing','NA', ...
    'SelectedVariableNames','UniqueCarrier');
% Convert MATLAB datastore to Spark RDD
rdd = sc.datastoreToRDD(ds);
```
В общем случае вход RDDs может быть создан с помощью следующих методов SparkContext класс: parallelize, datastoreToRDD, и textFile.
Выполните операции на RDD объект.
Используйте Spark метод RDD, такой как flatMap применять функцию ко всем элементам RDD возразите и сгладьте результаты. Функциональный carrierToCount это было создано, ранее служит функцией, которая будет прикладной к элементам RDD. Указатель на функцию к функциональному carrierToCount передается как входной параметр flatMap метод.
```
maprdd = rdd.flatMap(@carrierToCount);
redrdd = maprdd.reduceByKey( @(acc,value) acc+value );
countdata = redrdd.collect();

% Count and display carrier occurrences
count = 0;
for i=1:numel(countdata)
    count = count + countdata{i}{2};
    fprintf('\nCarrier Name: %s, Count: %d', countdata{i}{1}, countdata{i}{2});
end
fprintf('\n Total count : %d\n', count);

% Delete Spark Context
delete(sc)
```
В общем случае вы обеспечите указатели функций MATLAB или анонимные функции как входные параметры к Spark методы RDD, известные как transformations и actions. Эти указатели на функцию и анонимные функции выполняются на рабочих по развертываемому приложению.
Для списка поддерживаемых преобразований RDD и действий, смотрите Преобразования и Действия в разделе Methods RDD класс.
Для получения дополнительной информации о преобразованиях и действиях, смотрите Основы Spark Apache.
Создавание автономного приложения.
Используйте mcc команда с -m отметьте, чтобы создать автономное приложение. -m флаг создает стандартный исполняемый файл, который может быть запущен из командной строки. -a флаг включает зависимый набор данных airlinesmall.csv от папки <matlabroot>/toolbox/matlab/demos. mcc команда автоматически берет зависимый файл carrierToCount.m пока это находится в том же самом, работают папка.
```
>> mcc -m deployToSparkMlApiLocal.m -a <matlabroot>/toolbox/matlab/demos/airlinesmall.csv
```
mcc команда создает сценарий оболочки run_deployToSparkMlApiLocal.sh запускать исполняемый файл deployToSparkMlApiLocal.
Для получения дополнительной информации смотрите mcc.
Запустите автономное приложение от интерпретатора Linux используя следующую команду:
$ ./run_deployToSparkMlApiLocal.sh /share/MATLAB/MATLAB_Runtime/v91
/share/MATLAB/MATLAB_Runtime/v91 аргумент, указывающий на местоположение MATLAB Runtime.
До выполнения вышеупомянутой команды убедитесь javaclasspath.txt файл находится в той же папке как сценарий оболочки и исполняемый файл.
Ваше приложение не выполнится, если оно не может найти файл javaclasspath.txt.
Ваше приложение также может не выполниться, если дополнительная линия, содержащая местоположение папки конфигурационных файлов Hadoop, не прокомментирована. Выполнить ваше приложение на local кластерный менеджер, эта линия должна быть прокомментирована. Эта линия должна только быть не прокомментирована, если вы планируете выполнение вашего приложения с помощью yarn-client когда ваш кластерный менеджер на Spark включил кластер Hadoop.

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

Carrier Name: 9E, Count: 521
Carrier Name: AA, Count: 14930
Carrier Name: AQ, Count: 154
Carrier Name: AS, Count: 2910
Carrier Name: B6, Count: 806
Carrier Name: CO, Count: 8138
...
...
...
Carrier Name: US, Count: 13997
Carrier Name: WN, Count: 15931
Carrier Name: XE, Count: 2357
Carrier Name: YV, Count: 849
 Total count : 123523

Код:

deployToSparkMlApiLocal.m

Hadoop YARN

Менеджер по кластеру клиента пряжи представляет включенный кластер Hadoop Spark. Менеджер по кластеру YARN был введен в Hadoop 2.0. Это обычно устанавливается на тех же узлах как HDFS™. Поэтому под управлением Spark на YARN позволяет доступу к Spark данные HDFS легко. В приложениях это обозначается с помощью слова yarn-client.

Начиная с шагов для развертывания вашего приложения с помощью yarn-client как ваш кластерный менеджер похожи на использование локального кластерного менеджера, показанного выше, шагам дарят минимальное обсуждение. Для детального обсуждения каждого шага проверяйте Локальный случай выше.

Примечание

Можно следовать тем же инструкциям, чтобы развернуться, приложения Spark создали использование MATLAB API для 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 установлен.

Необходимые условия

Запустите этот пример путем создания новой папки работы, которая отображается к пути поиска файлов MATLAB.
Установите MATLAB Runtime в папке, которая доступна каждым узлом рабочего в кластере Hadoop. Этот пример использует /share/MATLAB/MATLAB_Runtime/v91 как местоположение папки MATLAB Runtime.
Если у вас нет MATLAB Runtime, можно загрузить его с веб-сайта в: https://www.mathworks.com/products/compiler/mcr.
Скопируйте 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','/share/hadoop/hadoop-2.6.0')
```
HADOOP_PREFIX переменная окружения должна быть установлена при использовании MATLAB datastore функционируйте, чтобы указать на данные по HDFS. Установка этой переменной окружения не имеет никакого отношения к Spark. Смотрите Отношение Между Spark и 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://hadoop01glnxa64:54310/user/<username>/sparkdeploy'});

Для получения дополнительной информации о свойствах Spark расширьте prop значение 'SparkProperties' пара "имя-значение" в разделе Input Arguments SparkConf класс.

Создайте SparkConf объект.
Используйте класс matlab.compiler.mlspark.SparkConf создать SparkConf объект.
```
conf = matlab.compiler.mlspark.SparkConf( ...
    'AppName','myApp', ...
    'Master','yarn-client', ...
    'SparkProperties',sparkProperties);
```
Для получения дополнительной информации о SparkConf смотрите matlab.compiler.mlspark.SparkConf.
Создайте SparkContext объект.
Используйте класс matlab.compiler.mlspark.SparkContext с SparkConf возразите как вход, чтобы создать SparkContext объект.
```
sc = matlab.compiler.mlspark.SparkContext(conf);
```
Для получения дополнительной информации о SparkContext смотрите matlab.compiler.mlspark.SparkContext.
Создайте RDD объект из данных.
Используйте функцию MATLAB datastore создать datastore объект, указывающий на файл airlinesmall.csv в HDFS. Затем используйте метод SparkContext datastoreToRDD преобразовывать datastore возразите против Spark RDD объект.
```
% Create a MATLAB datastore (HADOOP)
ds = datastore(...
    'hdfs:///user/<username>/datasets/airlinesmall.csv',...
    'TreatAsMissing','NA',...
    'SelectedVariableNames','UniqueCarrier');

% Convert MATLAB datastore to Spark RDD
rdd = sc.datastoreToRDD(ds);
```
В общем случае вход RDDs может быть создан с помощью следующих методов SparkContext класс: parallelize, datastoreToRDD, и textFile.
Выполните операции на RDD объект.
Используйте Spark метод RDD, такой как flatMap применять функцию ко всем элементам RDD возразите и сгладьте результаты. Функциональный carrierToCount это было создано, ранее служит функцией, которая будет прикладной к элементам RDD. Указатель на функцию к функциональному carrierToCount передается как входной параметр flatMap метод.
```
maprdd = rdd.flatMap(@carrierToCount);
redrdd = maprdd.reduceByKey( @(acc,value) acc+value );
countdata = redrdd.collect();

% Count and display carrier occurrences
count = 0;
for i=1:numel(countdata)
    count = count + countdata{i}{2};
    fprintf('\nCarrier Code: %s, Count: %d', countdata{i}{1}, countdata{i}{2});
end
fprintf('\n Total count : %d\n', count);

% Save results to MAT file
save('countdata.mat','countdata');

% Delete Spark Context
delete(sc);
```
Для списка поддерживаемых преобразований RDD и действий, смотрите Преобразования и Действия в разделе Methods RDD класс.
Для получения дополнительной информации о преобразованиях и действиях, смотрите Основы Spark Apache.
Создавание автономного приложения.
Используйте mcc команда с -m отметьте, чтобы создать автономное приложение. -m флаг создает автономное приложение, которое может быть запущено из командной строки. Вы не должны присоединять набор данных airlinesmall.csv поскольку это находится на HDFS. mcc команда автоматически берет зависимый файл carrierToCount.m пока это находится в том же самом, работают папка.
```
>> mcc -m deployToSparkMlApiHadoop.m
```
mcc команда создает сценарий оболочки run_deployToSparkMlApiHadoop.sh запускать исполняемый файл deployToSparkMlApiHadoop.
Для получения дополнительной информации смотрите mcc.
Запустите автономное приложение от интерпретатора Linux используя следующую команду:
$ ./run_deployToSparkMlApiHadoop.sh /share/MATLAB/MATLAB_Runtime/v91
/share/MATLAB/MATLAB_Runtime/v91 аргумент, указывающий на местоположение MATLAB Runtime.
До выполнения вышеупомянутой команды убедитесь javaclasspath.txt файл находится в той же папке как сценарий оболочки и исполняемый файл.
Ваше приложение не выполнится, если оно не может найти файл javaclasspath.txt.
Ваше приложение также может не выполниться, если дополнительная линия, содержащая местоположение папки конфигурационных файлов Hadoop, прокомментирована. Чтобы выполнить ваше приложение на менеджере по кластеру клиента пряжи, эта линия должна быть не прокомментирована. Эта линия должна только быть прокомментирована, если вы планируете выполнение вашего приложения с помощью локального кластерного менеджера.

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

Carrier Name: 9E, Count: 521
Carrier Name: AA, Count: 14930
Carrier Name: AQ, Count: 154
Carrier Name: AS, Count: 2910
Carrier Name: B6, Count: 806
Carrier Name: CO, Count: 8138
...
...
...
Carrier Name: US, Count: 13997
Carrier Name: WN, Count: 15931
Carrier Name: XE, Count: 2357
Carrier Name: YV, Count: 849
 Total count : 123523

Примечание

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

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

Например:

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

Код:

deployToSparkMlApiHadoop.m

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