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

%   Copyright 2022-2024 The MathWorks, Inc.

classdef ParallelSparkMapReducer < matlab.mapreduce.MapReducer

    properties
        % The parallel.cluster.Spark object to be used by this MapReducer.
        Cluster;
    end

    methods
        % Construct an MapReducer that executes mapreduce on the given Spark cluster.
        function obj = ParallelSparkMapReducer(cluster)
            if nargin < 1
                cluster = parallel.cluster.Spark;
            end
            obj.Cluster = cluster;
        end

        function set.Cluster(obj, newValue)
            validateattributes(newValue, {'parallel.cluster.Spark'}, {'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) && hIsSparkClusterValid(obj.Cluster);
        end

        % Hidden execute method for mapreduce
        function out = execMapReduce(obj, inputds, mapFunction, reduceFunction, settings)
            % We do not support any datastore that does not support
            % splitting.
            envName = getString(message('parallel:bigdata:SparkEnvironmentName'));
            matlab.io.datastore.internal.validateParallelBigdata(...
                inputds, ["Partitionable", "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:SparkRequiresOutputFolder'));
            end

            executor = obj.getPartitionedArrayExecutor();
            settings = parallel.internal.mapreduce.Settings(settings);
            out = parallel.internal.bigdata.sparkMapreduce(...
                executor, inputds, mapFunction, reduceFunction, settings);
        end
    end

    methods (Access = protected)
        function executor = createPartitionedArrayExecutor(obj)
            import parallel.internal.hadoop.HadoopSparkJobInfo;

            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 ParallelSparkMapReducer 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 Spark
            % cluster properties are modified. This copy is never used
            % outside of equality checks.
            cluster = obj.Cluster;
            clusterProps = hGetMementoData(cluster, parallel.internal.types.ClusterMementoType.Full);
            m.ClusterHash = distcompserialize(clusterProps);
        end
    end

    methods (Access = protected, Static)
        % Recreate a ParallelSparkMapReducer object from a memento struct
        function obj = doCreateFromMemento(m)
            obj = matlab.mapreduce.ParallelSparkMapReducer(m.Cluster);
            obj.ObjectVisibility = m.ObjectVisibility;
        end
    end
end

