%HadoopSparkJob
% Internal class that represents a Hadoop Spark Job in the PCT framework.
%
% This currently bends the definition of parallel.Job because the business
% logic of SparkContext is combined with the underlying scheduling process.
% This means this class is more of a internal wrapper than a true Job
% class.
%

%   Copyright 2016-2021 The MathWorks, Inc.

classdef (Sealed) HadoopSparkJob < parallel.Job & parallel.internal.hadoop.HadoopJobMixin
    
    properties (SetAccess = private)
        % An implementation of SparkIntegContext that represents how to
        % access the running Spark Job.
        SparkIntegContext;
    end
    
    methods
        % Construct a HadoopSparkJob from the given cluster and construction
        % arguments for the info object.
        function obj = HadoopSparkJob(cluster, info)
            obj = obj@parallel.Job(cluster, 0, parallel.internal.types.Variant.IndependentJob);
            obj = obj@parallel.internal.hadoop.HadoopJobMixin(info);
        end
    end
    
    methods (Access = protected)
        % getTasks must return a list of parallel.Task objects
        function tl = getTasks(~)
            tl = parallel.Task.empty();
        end
        
        % getTaskCount return the number of tasks
        function tc = getTaskCount( obj )
            tc = numel(obj.getTasks);
        end
        
        % Submit a scalar job. This will currently block until the Spark
        % job is in a running state.
        function doSubmit(obj, workerDataFiles)
            import parallel.internal.hadoop.HadoopWorker;
            import parallel.internal.mapreduce.TempLocalFolder;
            
            cluster = obj.Parent;
            
            sparkInstallFolder = cluster.SparkInstallFolder;
            if isempty(sparkInstallFolder) || exist(sparkInstallFolder, 'dir') ~= 7
                error(message('parallel:mapreduce:InvalidSparkInstallFolder', sparkInstallFolder));
            end
            
            % We need to pass configurations about how to launch a MATLAB
            % Worker through the Hadoop configuration object passed through
            % Spark.
            if isempty(cluster.HadoopProperties.keys)
                hadoopConfiguration = containers.Map('KeyType', 'char', 'ValueType', 'char');
            else
                hadoopConfiguration = containers.Map(cluster.HadoopProperties.keys, cluster.HadoopProperties.values);
            end
            data = obj.getMatlabWorkerEnvironmentOptions();
            for ii = 1:size(data, 1)
                hadoopConfiguration(data{ii, 1}) = data{ii, 2};
            end
            
            if isempty(cluster.SparkProperties.keys)
                sparkConfiguration = containers.Map('KeyType', 'char', 'ValueType', 'char');
            else
                sparkConfiguration = containers.Map(cluster.SparkProperties.keys, cluster.SparkProperties.values);
            end
            
            % There are several options that we can only set at the client
            % that require the cluster architecture. To work around this,
            % we hedge our bets and place options for all supported architectures.
            % Note, Spark will convert <CPS> to the path separator of the
            % worker.
            sparkConfiguration('spark.executor.extraLibraryPath') = strjoin({...
                obj.getWorkerLibraryPath('glnxa64', '<CPS>'), ...
                obj.getWorkerLibraryPath('maci64', '<CPS>'), ...
                obj.getWorkerLibraryPath('win64', '<CPS>')}, '<CPS>');
            
            try
                obj.SparkIntegContext = parallel.internal.bigdata.MdcsSparkIntegContext(...
                    'HadoopInstallFolder',  cluster.HadoopInstallFolder, ...
                    'SparkInstallFolder', sparkInstallFolder, ...
                    'AttachedFiles', workerDataFiles, ...
                    'HadoopConfiguration', hadoopConfiguration, ...
                    'SparkConfiguration', sparkConfiguration);
            catch err
                throw(addCause(MException(message('parallel:mapreduce:HadoopSparkJobSubmitFailure', sparkInstallFolder)), err));
            end
        end
    end
end
