%ParallelHadoopMapReducer
% matlab.mapreduce.ParallelHadoopMapReducer is the MapReducer class that
% represents executing mapreduce on a Hadoop cluster with MATLAB workers.
%
%   matlab.mapreduce.ParallelHadoopMapReducer properties:
%      Cluster - The parallel.cluster.Hadoop object to be used by this MapReducer.
%
% See also mapreducer, parallel.cluster.Hadoop.

%   Copyright 2014-2020 The MathWorks, Inc.

classdef ParallelHadoopMapReducer < matlab.mapreduce.MapReducer
    
    properties
        Cluster; % The parallel.cluster.Hadoop object to be used by this MapReducer.
    end
    
    methods
        % Construct an MapReducer that executes mapreduce on the given Hadoop cluster.
        function obj = ParallelHadoopMapReducer(cluster)
            if nargin < 1
                cluster = parallel.cluster.Hadoop;
            end
            obj.Cluster = cluster;
        end
        
        function set.Cluster(obj, newValue)
            validateattributes(newValue, {'parallel.cluster.Hadoop'}, {'scalar'});
            obj.Cluster = newValue;
        end
    end
    
    methods (Hidden)
        % Check if this MapReducer is valid to be used as the current MapReducer.
        function isValid = isValidForCurrentMapReducer(obj)
            isValid = strcmp(obj.ObjectVisibility, 'On') && isvalid(obj.Cluster) && hIsHadoopClusterValid(obj.Cluster);
        end
        
        % Hidden execute method for mapreduce
        function out = execMapReduce(obj, inputds, mapFunction, reduceFunction, settings)
            import parallel.internal.mapreduce.Settings;
            import parallel.internal.hadoop.HadoopMapReduceJob;
            import parallel.internal.hadoop.HadoopMapReduceJobInfo;
            
            if ~hIsHadoopClusterValid(obj.Cluster)
                error(message('parallel:mapreduce:HadoopClusterObjectNotValid'));
            end
            
            % We do not support any datastore that does not support
            % splitting.
            envName = getString(message('MATLAB:bigdata:executor:HadoopEnvironmentName'));
            matlab.io.datastore.internal.validateParallelBigdata(...
                inputds, "HadoopLocationBased", envName);
            
            % The default for OutputFolder is the current directory. As
            % Hadoop requires the output folder to be non-existent, this
            % will typically error so we guard here against it.
            if isempty(settings.OutputFolder)
                error(message('parallel:mapreduce:HadoopRequiresOutputFolder'));
            end
            
            % If the Hadoop has not already been setup on the client
            % MATLAB, do so here.
            iSetupClient(obj.Cluster.HadoopInstallFolder);
            
            % A flag that obtains the old display behavior.
            useLogDisplay = distcomp.feature('HadoopUseLogDisplay');
            
            try
                settings = Settings(settings);
                if strcmp(settings.Display, 'On')
                    disp(getString(message('parallel:mapreduce:ParallelHadoopMapReducerDisplayHeader')));
                end
                
                mrprogress = matlab.mapreduce.internal.MapReduceProgress(strcmp(settings.Display, 'On'), 1);
                mrprogress.PrintHeader();
                if ~useLogDisplay
                    mrprogress.PrintZero();
                end
                
                clusterImpl = obj.Cluster.Impl;
                jobInfo = HadoopMapReduceJobInfo(inputds, mapFunction, reduceFunction, settings);
                job = jobInfo.wrapAsJob(clusterImpl);
                job.AutoAttachFiles = clusterImpl.AutoAttachFiles;
                job.AttachedFiles = clusterImpl.AttachedFiles;
                job.AdditionalPaths = clusterImpl.AdditionalPaths;
                if ~useLogDisplay
                    job.DisplayObject = mrprogress;
                end
                
                
                % This submit will block until the job has completed.
                job.submit();
                
                if ~useLogDisplay
                    mrprogress.PrintHundred();
                end
                
                out = iCreateOutputDatastore(settings.getOutputFolderAsIri(), settings.OutputType);
            catch err
                err.throw();
            end
        end
    end
    
    methods (Access = protected)
        function executor = createPartitionedArrayExecutor(obj)
            import parallel.internal.hadoop.HadoopSparkJobInfo;
            
            if ~hIsHadoopClusterValid(obj.Cluster)
                error(message('parallel:mapreduce:HadoopClusterObjectNotValid'));
            elseif obj.Cluster.Impl.HadoopVersion < 2
                error(message('parallel:mapreduce:HadoopSparkUnsupportedVersion'));
            end
            
            % If the Hadoop has not already been setup on the client
            % MATLAB, do so here.
            iSetupClient(obj.Cluster.HadoopInstallFolder);
            
            fprintf('%s', getString(message('parallel:mapreduce:StartingSparkContext')));
            
            % As the StartingSparkContextDone message is intended to be
            % printed on the same line as the StartingSparkContext message,
            % we defer the new line until the end of this function. If an
            % error occurs, this will ensure that error is printed to the
            % next line.
            printCleanup = onCleanup(@()fprintf('\n'));
            
            jobInfo = HadoopSparkJobInfo();
            job = jobInfo.wrapAsJob(obj.Cluster.Impl);
            job.AutoAttachFiles = obj.Cluster.Impl.AutoAttachFiles;
            job.AttachedFiles = obj.Cluster.Impl.AttachedFiles;
            job.AdditionalPaths = obj.Cluster.Impl.AdditionalPaths;
            submit(job);
            
            executor = matlab.bigdata.internal.spark.SparkExecutor(job.SparkIntegContext);
            
            fprintf('%s', getString(message('parallel:mapreduce:StartingSparkContextDone')));
        end
        
        % Create a memento struct from a ParallelHadoopMapReducer object.
        function m = doGetMemento(obj)
            m = doGetMemento@matlab.mapreduce.MapReducer(obj);
            
            % We include a serialized copy of the cluster properties so
            % that the memento is guaranteed to change if any Hadoop
            % cluster properties are modified. This copy is never used
            % outside of equality checks.
            cluster = obj.Cluster;
            propnames = properties(cluster);
            for ii = 1:numel(propnames)
                clusterProps.(propnames{ii}) = cluster.(propnames{ii});
            end
            m.ClusterHash = distcompserialize(clusterProps);
        end
    end
    
    methods (Access = protected, Static)
        % Recreate a ParallelHadoopMapReducer object from a memento struct
        function obj = doCreateFromMemento(m)
            obj = matlab.mapreduce.ParallelHadoopMapReducer(m.Cluster);
            obj.ObjectVisibility = m.ObjectVisibility;
        end
    end
end

function iSetupClient(installPath)
hadoopHome = getenv('HADOOP_HOME');
hadoopEnvCleanup = onCleanup(@()setenv('HADOOP_HOME', hadoopHome));
setenv('HADOOP_HOME', installPath);
matlab.io.internal.vfs.hadoop.hadoopLoader;
end

% Create a datastore that represents the output.
function ds = iCreateOutputDatastore(outputFolder, outputType)
switch outputType
    case 'Binary'
        files = iListOutputFiles(outputFolder, '.seq');
        ds = datastore(files, 'Type', 'KeyValue', 'FileType', 'seq');
    case 'TabularText'
        files = iListOutputFiles(outputFolder, '.txt');
        ds = matlab.mapreduce.internal.createTextDSWithKeyValueVarNames(files);
    otherwise
        assert(false)
end
end

% Helper function that lists the output files.
function files = iListOutputFiles(outputFolder, requiredExtension)
import matlab.io.datastore.internal.HadoopConfiguration;
import parallel.internal.hadoop.FileSystemTools

fileSystemTools = FileSystemTools(HadoopConfiguration.getGlobalConfiguration());
[files, isFileEmpty] = fileSystemTools.listFilesInFolder(outputFolder);

[~, ~, ext] = cellfun(@fileparts, files, 'UniformOutput', false);
isMatchingExt = strcmp(ext, requiredExtension);
files = files(isMatchingExt);
isFileEmpty = isFileEmpty(isMatchingExt);

if isempty(files)
    error(message('parallel:mapreduce:HadoopOutputNotFound', outputFolder));
end

% We remove files that are empty as format introspection fails for
% TabularTextDatastore if the first file is empty.
files = files(~isFileEmpty);
end
