%DeployHadoopMapReducer
% matlab.mapreduce.DeployHadoopMapReducer is the MapReducer class that
% represents executing mapreduce on a Hadoop cluster with deployed MCR.
%
%   matlab.mapreduce.DeployHadoopMapReducer properties:
%     HadoopInstallFolder - Hadoop installation folder to be used.
%     HadoopConfigurationFile - Hadoop cluster configuration file
%     MCRRoot - MCR installation folder for Hadoop cluster
%     HadoopProperties - Hadoop cluster properties
%
% See also mapreducer.

%   Copyright 2014-2020 The MathWorks, Inc.

classdef DeployHadoopMapReducer < matlab.mapreduce.MapReducer
%#function matlab.compiler.mlhadoop.internal.mapwrapper
%#function matlab.compiler.mlhadoop.internal.reducewrapper
%#function compiler.internal.genHadoopConfig

    properties (Access = public)
        hadoopInstallFolder
        hadoopCfgFile
        hadoopProperties
        mcrRoot
    end
    
    methods (Access = public)
        function obj = DeployHadoopMapReducer(varargin)
            p = inputParser;
            defaultMCRRoot = matlabroot;
            %default Hadoop install folder if not specified by user
            defaultHadoopInstallFolder = matlab.io.internal.vfs.hadoop.discoverHadoopInstallFolder();
            defaultHadoopConfigurationFile = '';
            
            % Parse inputs, validate data type, not file existance (yet)
            addParameter(p,'HadoopInstallFolder', ...
                defaultHadoopInstallFolder,@(x)validateattributes(x, ...
                {'char','string'},{'scalartext'}, ...
                'matlab.mapreduce.DeployHadoopMapReducer', ...
                'HadoopInstallFolder'))
            addParameter(p,'HadoopConfigurationFile', ...
                defaultHadoopConfigurationFile, ...
                @(x)validateattributes(x, ...
                {'char','string'},{'scalartext'}, ...
                'matlab.mapreduce.DeployHadoopMapReducer', ...
                'HadoopConfigurationFile'))
            addParameter(p,'MCRRoot',defaultMCRRoot, ...
                @(x)validateattributes(x, ...
                {'char','string'},{'scalartext'}, ...
                'matlab.mapreduce.DeployHadoopMapReducer','MCRRoot'))
            addParameter(p,'HadoopProperties',containers.Map('KeyType','char','ValueType','char'), ...
                @(x)isa(x,'containers.Map') && strcmp(x.KeyType,'char') && strcmp(x.ValueType,'char'));
            
            parse(p, varargin{:});
            
            % Check for file existance of HadoopInstallFolder and
            % HadoopConfigurationFile
            if ~exist(char(p.Results.HadoopInstallFolder),'dir')
                error('Bad value for HadoopInstallFolder. The following directory does not exist:\n\n%s',p.Results.HadoopInstallFolder)
            end
            if ~(any(strcmp('HadoopConfigurationFile',p.UsingDefaults)) || ...
                    exist(char(p.Results.HadoopConfigurationFile),'file'))
                error('Bad value for HadoopConfigurationFile. The following file does not exist:\n\n%s',p.Results.HadoopConfigurationFile)
            end
            
            obj.hadoopInstallFolder = char(p.Results.HadoopInstallFolder);
            obj.hadoopCfgFile = char(p.Results.HadoopConfigurationFile);
            obj.hadoopProperties = p.Results.HadoopProperties;
            obj.mcrRoot = char(p.Results.MCRRoot);
            
        end
    end
    
    methods (Hidden = true)
        function outputds = execMapReduce(this, ds, mapFcn, redFcn, parsedStruct) %#ok<INUSL>
            import matlab.mapreduce.internal.*;
            import matlab.mapreduce.*;
            
            if ~isdeployed
                error('DeployHadoopMapReducer should only be used in deployed mode.');
            end
            % Set up Hadoop environment variable so that hadoopLoader will work.
            setenv('MATLAB_HADOOP_INSTALL', this.hadoopInstallFolder);
            outputType = parsedStruct.OutputType;
            outputFolder = parsedStruct.OutputFolder;
            if (isempty(outputFolder))
                error('Mapreduce using HADOOP requires that the ''OutputFolder'' parameter is specified and set to a path that does not exist.');
            end
            
            % MapReduce Progress object
            mrprogress = matlab.mapreduce.internal.MapReduceProgress(true);
            
            % Display progress header
            mrprogress.PrintHeader();
            matfile = tempname;
            matfile = [matfile,'.mat'];
            [~,matfilename,~] = fileparts(matfile);
            if isa(ds,'matlab.io.datastore.KeyValueDatastore') && ~strcmp(ds.FileType,'seq')
                whos ds;
                error('Input type is not supported');
            end
            save(matfile, 'ds','mapFcn', 'redFcn');
            
            kvlimit = 1;
            if isa(ds, 'matlab.io.datastore.KeyValueDatastore')
                kvlimit = getKeyValueLimit(ds);
            end
            
            % generate hadoop configuration properties file
            configcell = compiler.internal.genHadoopConfig(matfile);
            configfile = tempname;
            fid = fopen(configfile,'w');
            fprintf(fid,'%s=%s\n',configcell{:});
            fprintf(fid,'mw.mcrroot=%s\n', this.mcrRoot );
            fprintf(fid,'mw.ds.out.type=%s\n', outputType);
            fprintf(fid,'mw.appmemento=%s.mat\n', matfilename);
            fclose(fid);
            
            C1 = onCleanup(@()delete(matfile));
            C2 = onCleanup(@()delete(configfile));
            
            % build hadoop command
            %1. hadoop
            %2. matlabroot to build mwmapreduce.jar path, mw.libpath
            %3. options
            %4. CTF file.
            %5. input files
            %6. output folder
            
            
            CTFFileName = getmcruserdata('CTFFileNameForHadoop');
            if (isempty(CTFFileName))
                error('Missing CTF file for application');
            end
            
            InputFile = matlab.io.datastore.internal.shim.getUnresolvedLocation(ds);
            if (iscell(InputFile))
                InputFile=strjoin(InputFile,',');
            end
            ARCHSTR=computer('arch');
            % Use Hadoop install folder to find out hadoop version and dependent JAR files.
            hadoopVersion = matlab.io.internal.vfs.hadoop.discoverHadoopVersion(this.hadoopInstallFolder);
            majorVersionNumber = str2double(regexp(hadoopVersion, '^\d', 'match', 'once'));
            
            if majorVersionNumber == 1
                error('Hadoop version %s is not supported. You must upgrade to Hadoop version 2 or later.', hadoopVersion);
            elseif (majorVersionNumber == 2)  || (majorVersionNumber == 3)
                strver = 'a2.2.0';
            else
                error('Hadoop version %s is not supported.', hadoopVersion);
            end
            
            if exist(fullfile(matlabroot,'toolbox/mlhadoop/jar', strver,'mwmapreduce.jar'),'file')
                strcompositejar = fullfile(matlabroot,'toolbox/mlhadoop/jar', strver,'mwmapreduce.jar');
            elseif exist(fullfile(matlabroot,'toolbox/mlhadoop/jar', strver, ARCHSTR , 'mwmapreduce.jar'),'file')
                strcompositejar = fullfile(matlabroot,'toolbox/mlhadoop/jar', strver, ARCHSTR , 'mwmapreduce.jar');
            else
                error('mwmapreduce.jar file does not exist. Please check MCR installation.');
            end
            hadoopOptions = '';
            if (~isempty(this.hadoopCfgFile))
                hadoopOptions = ['-conf "' this.hadoopCfgFile '" '];
            end
            if (~isempty(this.hadoopProperties))
                listOfPropNames = this.hadoopProperties.keys;
                for i = 1:length(listOfPropNames)
                    hadoopOptions = [hadoopOptions '-D ' ...
                        listOfPropNames{i} '=' this.hadoopProperties(listOfPropNames{i}) ' ']; %#ok<AGROW>
                end
            end
            
            HadoopCommand = fullfile(this.hadoopInstallFolder,'bin/hadoop');
            HadoopCommand = ['"' HadoopCommand '" jar "'  strcompositejar '"'...
                             ' com.mathworks.hadoop.MWMapReduceDriver ' hadoopOptions ...
                             ' -D mw.mode=application' ...
                             ' -D settingfile="' configfile '" '  ...
                             ' -files "' matfile '" ' ...
                             CTFFileName ' "'  InputFile '" "' outputFolder '"'];
            
            disp('executing:') ;
            disp(HadoopCommand);
            retStatus = system(HadoopCommand);
            if (retStatus == 0)
                if (strcmpi(outputType,'tabulartext'))
                    outputds = matlab.mapreduce.internal.createTextDSWithKeyValueVarNames(fullfile(outputFolder ,'part*'));
                else
                    % binary, seq file
                    outputds =  datastore(fullfile(outputFolder ,'part*'),'Type', 'KeyValue', 'FileType','seq','ReadSize',kvlimit );
                end
            else
                error('Hadoop job execution error: %d.', retStatus);
            end
        end
    end
    
    methods (Access = protected, Static)
        % Recreate a DeployHadoopMapReducer object from a memento struct.
        function obj = doCreateFromMemento(m)
            inputs = struct(...
                'HadoopInstallFolder', {m.hadoopInstallFolder}, ...
                'MCRRoot', {m.mcrRoot}, ...
                'HadoopProperties', {m.hadoopProperties});
            if ~isempty(m.hadoopCfgFile)
                inputs.HadoopConfigurationFile = m.hadoopCfgFile;
            end
            obj = matlab.mapreduce.DeployHadoopMapReducer(inputs);
            obj.ObjectVisibility = m.ObjectVisibility;
        end
    end
end
