classdef DeploySparkMapReducer < matlab.mapreduce.MapReducer
    %#function deserializeHelper
    %#function serializeHelper
    %#function callWithStringSerializedData
    %#function matlab.bigdata.internal.spark.worker.getWorkerPid
    %#function matlab.bigdata.internal.spark.worker.IteratorWrapper.create
    
    % matlab.mapreduce.DeploySparkMapReducer is the Spark class that
    % represents executing tall with Spark on a cluster with deployed MCR.
    %
    % See also mapreducer.
    %   Copyright 2016-2018 The MathWorks, Inc.
    properties (SetAccess = immutable)
        
        % The Spark application name.
        sparkAppName;
        
        % The cluster manager to connect to.
        sparkMaster;
        
        % Main entry point for Spark functionality, it represents the 
        % connection to a Spark cluster.
        sparkContext;
        
        % A map of key-value pairs that will be passed to Spark to set its 
        % properties.
        sparkProperties;
        
        % Matlab Runtime installation folder on the cluster.
        mcrRoot;
        
        %Spark log level
        sparkLogLevel;
    end
    
    properties(SetAccess = immutable, Hidden = true)
        % The installation directory of Spark.
        sparkInstallFolder;
    end
    
    methods
         function obj = DeploySparkMapReducer(varargin)
            
            if ~isdeployed || ~strcmpi(computer, 'GLNXA64')
                error(message('compiler_tall:internal:UnsupportedMatlabMode'));
            end
            
            matlab.mapreduce.DeploySparkMapReducer.setUpSparkClassPath();
            p = inputParser;
            
            % Need to ensure Spark can see the dynamic classpath, this is
            % done by setting the thread context classLoader during
            % initialization.
            cls = com.mathworks.util.ClassLoaderBridge.findClass(...
                "com.mathworks.toolbox.shared.bigdata.spark.SparkIntegContext");
            oldClassLoader = java.lang.Thread.currentThread.getContextClassLoader();
            thread = java.lang.Thread.currentThread();
            oldClassLoaderCleanup = onCleanup(@() thread.setContextClassLoader(oldClassLoader));
            thread.setContextClassLoader(cls.getClassLoader);

            defaultAppName = '';
            defaultMaster = '';
            defaultMCRRoot = matlabroot;
            defaultSparkLogLevel = 'INFO';

            addParameter(p, 'AppName', defaultAppName, ...
                @(x)validateattributes(x, {'char', 'string'}, {'scalartext'}));
            addParameter(p, 'Master',  defaultMaster, ...
                @(x)validateattributes(x, {'char', 'string'}, {'scalartext'}));
            addParameter(p, 'MCRRoot', defaultMCRRoot, ...
                @(x)validateattributes(x, {'char', 'string'}, {'scalartext'}));
            addParameter(p, 'SparkLogLevel', defaultSparkLogLevel, ...
                @(x)validateattributes(x, {'char', 'string'}, {'scalartext'}));
            addParameter(p, 'SparkProperties',containers.Map('KeyType','char','ValueType','char'), ...
                @(x)isa(x,'containers.Map') && strcmp(x.KeyType,'char') && strcmp(x.ValueType,'char'));
            
            parse(p, varargin{:});

            obj.sparkMaster = char(p.Results.Master);
            % if Master is local, local[*], local[#], error out
            if ~isempty(regexp(obj.sparkMaster,'^local(\[((\d+)|\*)\])?$','match'))
                error(message('compiler_tall:internal:UnsupportedSparkMode'));
            end
            
            obj.sparkAppName = char(p.Results.AppName);
            obj.sparkProperties = p.Results.SparkProperties;
            obj.mcrRoot = char(p.Results.MCRRoot);
            obj.sparkLogLevel = char(p.Results.SparkLogLevel);
            
            obj.sparkInstallFolder = getenv('SPARK_HOME');
            sparkConf = org.apache.spark.SparkConf;
            
            if ~isempty(obj.sparkAppName)
                sparkConf.setAppName(obj.sparkAppName);
            end

            if ~isempty(obj.sparkMaster)
                sparkConf.setMaster(obj.sparkMaster);
            end
            
            % Disable this check by default because we have explicit errors
            % against out of memory that should be more precise.
            sparkConf.set('spark.driver.maxResultSize', '0');
            
            if (~isempty(obj.sparkInstallFolder))
                sparkConf.setSparkHome(obj.sparkInstallFolder);            
            end
            
            if (~isempty(obj.sparkProperties))
               listOfPropNames = obj.sparkProperties.keys;
               for i = 1:length(listOfPropNames)
                   sparkConf.set(listOfPropNames{i}, obj.sparkProperties(listOfPropNames{i}));
               end
            end

            sparkConf.set('spark.executorEnv.LD_LIBRARY_PATH', ...
                strjoin({fullfile(obj.mcrRoot, 'runtime', 'glnxa64'), ...
                         fullfile(obj.mcrRoot, 'bin', 'glnxa64'), ...
                         fullfile(obj.mcrRoot, 'sys', 'os', 'glnxa64'), ...
                         fullfile(obj.mcrRoot, 'sys', 'opengl', 'lib', 'glnxa64')}, ':'));
            
            ctfFileName = getmcruserdata('CTFFileNameForHadoop');
            
            if (~exist(ctfFileName,'file') )
                % If CTF file name is empty, pass empty string
                if isempty(ctfFileName)
                    ctfFileName = '';
                end
                error(message('compiler_tall:internal:CTFFileDoesNotExist', ctfFileName));
            end
            
            [~, name, ext] = fileparts(ctfFileName);
            sparkConf.set('spark.executorEnv.mathworksCTFFileName', [name ext]);
            
            % Create the Java spark context
            obj.sparkContext = org.apache.spark.api.java.JavaSparkContext(sparkConf);
            
            % If the the master is set as local, local[*], local[#] in the conf file or spark-submit
            if ~isempty(regexp(char(obj.sparkContext.master()),'^local(\[((\d+)|\*)\])?$','match'))
                error(message('compiler_tall:internal:UnsupportedSparkMode'));
            end

            % Set Log level
            if ~strcmp(obj.sparkLogLevel,defaultSparkLogLevel)
                sparkVer = obj.sparkContext.version();
                vercell = regexp(char(sparkVer),'(\d.\d).\d','tokens');
                if ~isempty(vercell) && strcmp(vercell{1}{1},'1.3')
                    warning(message('compiler_tall:internal:UnsupportedSparkAPI'));
                else
                    obj.sparkContext.setLogLevel(obj.sparkLogLevel);
                end
            end
            
            % pass the ctf file        
            obj.sparkContext.addFile(ctfFileName);
            
            archStr = computer('arch');
            
            % While running against MATLAB v/s MCR, the location of 
            % mwmapreduce.jar is different
            mwmapreducejar = fullfile(matlabroot, 'toolbox', 'mlhadoop', 'jar', 'a2.2.0', 'mwmapreduce.jar');
            if ~exist(mwmapreducejar, 'file')
                mwmapreducejar = fullfile(matlabroot, 'toolbox', 'mlhadoop', 'jar', 'a2.2.0', archStr, 'mwmapreduce.jar');
            end

            SupportingJarFiles = {...
                fullfile(matlabroot, 'toolbox', 'compiler', 'mltall', 'buildmcr', 'deployedworkers', 'deployedworkers.jar'), ...
                mwmapreducejar, ...
                fullfile(matlabroot, 'java', 'jar', 'toolbox', 'shared', 'bigdata.jar'), ...
                fullfile(matlabroot, 'toolbox', 'shared', 'bigdata', 'jar', 'hadoop.2.jar'), ...
                fullfile(matlabroot, 'toolbox', 'shared', 'bigdata', 'jar', 'sparkinteg.jar'), ...
                };

            for j = 1:numel(SupportingJarFiles)
                obj.sparkContext.addJar(SupportingJarFiles{j});
            end
            
        end
        
        function delete(obj)
            obj.sparkContext.stop();
        end
        
    end

    methods(Access = protected)
        function executor = createPartitionedArrayExecutor(obj)
            context = matlab.bigdata.internal.spark.InProcessSparkIntegContext(obj.sparkContext, ...
                'MatlabWorkerFactoryClassname', 'com.mathworks.toolbox.compiler.tall.InProcessDeployedWorkerFactory', ...
                'SerializeFunctionName', 'serializeHelper', ...
                'DeserializeFunctionName', 'deserializeHelper', ...
                'UseSingleWorker', true );
            
            executor = matlab.bigdata.internal.spark.SparkExecutor(context);
        end
    end
    
    methods(Hidden = true)
        function execMapReduce(~)
            error(message('compiler_tall:internal:UnsupportedMapReduceFunction'));
        end
    end
    
    methods(Static, Hidden = true)
        function setUpSparkClassPath()
            persistent jarsAdded
            if isempty(jarsAdded)
                % jars may be present on the static classpath
                oldWarnState = warning('off', 'MATLAB:javaclasspath:jarAlreadySpecified');
                warnCleanup = onCleanup(@()warning(oldWarnState));
                javaaddpath(fullfile(matlabroot, 'toolbox', 'shared', 'bigdata', 'jar', 'sparkinteg.jar')); %SparkIntegContext
                jarsAdded = true;
            end
        end
    end
    
    methods (Access = protected)
        % Create a memento struct that can recreate this DeploySparkMapReducer.
        function m = doGetMemento(obj)
            m = doGetMemento@matlab.mapreduce.MapReducer(obj);
            % We do not want to the lifetime of the Spark context to be
            % tied to the memento struct.
            m = rmfield(m, {'sparkContext'});
        end
    end
    
    methods (Access = protected, Static)
        % Recreate a DeploySparkMapReducer object from a memento struct.
        function obj = doCreateFromMemento(m)
            inputs = struct(...
                'AppName', m.sparkAppName, ...
                'Master', m.sparkMaster, ...
                'SparkProperties', m.sparkProperties, ...
                'MCRRoot', m.mcrRoot, ...
                'SparkLogLevel', m.sparkLogLevel);
            obj = matlab.mapreduce.DeploySparkMapReducer(inputs);
            obj.ObjectVisibility = m.ObjectVisibility;
        end
    end    
end
