<commit_before>package org.apache.hadoop.tools.posum.simulator.predictor;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapreduce.v2.api.records.TaskType;
import org.apache.hadoop.tools.posum.common.records.dataentity.DataEntityType;
import org.apache.hadoop.tools.posum.common.records.dataentity.JobProfile;
import org.apache.hadoop.tools.posum.common.records.dataentity.TaskProfile;
import org.apache.hadoop.tools.posum.common.util.POSUMConfiguration;
import org.apache.hadoop.tools.posum.common.util.POSUMException;

import java.util.List;

/**
 * Created by ane on 2/9/16.
 */
public class StandardPredictor extends JobBehaviorPredictor {

    private static final Log logger = LogFactory.getLog(StandardPredictor.class);

    private List<JobProfile> getComparableProfiles(JobProfile job, TaskType type) {
        int bufferLimit = conf.getInt(POSUMConfiguration.PREDICTION_BUFFER,
                POSUMConfiguration.PREDICTION_BUFFER_DEFAULT);
        // get past jobs with the same name
        List<JobProfile> comparable = getDataStore().find(
                DataEntityType.JOB_HISTORY,
                type.equals(TaskType.MAP) ? "mapperClass" : "reducerClass",
                type.equals(TaskType.MAP) ? job.getMapperClass() : job.getReducerClass(),
                -bufferLimit,
                bufferLimit
        );
        if (comparable.size() < 1) {
            // get past jobs at least by the same user
            comparable = getDataStore().find(
                    DataEntityType.JOB_HISTORY,
                    "user",
                    job.getUser(),
                    -bufferLimit,
                    bufferLimit
            );
        }
        return comparable;
    }

    @Override
    public Long predictJobDuration(String jobId) {
        JobProfile job = getDataStore().findById(DataEntityType.JOB, jobId);
        return predictMapTaskDuration(job) * job.getTotalMapTasks() +
                predictReduceTaskDuration(job) * job.getTotalReduceTasks();
    }

    @Override
    public Long predictLocalMapTaskDuration(String jobId) {
        return predictTaskDuration(jobId, TaskType.MAP);
    }


    private Long predictMapTaskDuration(JobProfile job) {
        if (job.getAvgMapDuration() != 0)
            return job.getAvgMapDuration();
        // we have no information about this job; predict from history
        List<JobProfile> comparable = getComparableProfiles(job, TaskType.MAP);
        if (comparable.size() < 1) {
            logger.debug("No map history data for " + job.getId() + ". Using default");
            return conf.getLong(POSUMConfiguration.AVERAGE_TASK_DURATION,
                    POSUMConfiguration.AVERAGE_TASK_DURATION_DEFAULT);
        }
        double avgMapRate = 0;
        for (JobProfile profile : comparable) {
            //restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
            avgMapRate += 1.0 * Math.max(profile.getTotalInputBytes(), profile.getTotalMapTasks()) / profile.getAvgMapDuration();
        }
        avgMapRate /= comparable.size();
        // restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
        Double duration = Math.max(job.getTotalInputBytes(), job.getTotalMapTasks()) / avgMapRate;
        logger.debug("Map duration for " + job.getId() + " should be " + job.getTotalInputBytes() + " / " + avgMapRate + " = " + duration);
        return duration.longValue();
    }

    private double calculateMapTaskSelectivity(JobProfile job) {
        // we try to compute selectivity from the map history
        List<JobProfile> comparable = getComparableProfiles(job, TaskType.MAP);
        if (comparable.size() < 1 || !comparable.get(0).getMapperClass().equals(job.getMapperClass())) {
            // there is no history, or it is not relevant for selectivity
            if (job.getCompletedMaps() > 0)
                // we know the current selectivity
                // restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
                return job.getMapOutputBytes() / Math.max(1.0 * job.getTotalInputBytes() / job.getCompletedMaps(), job.getCompletedMaps());
            // we don't know anything about selectivity
            return 0;
        }
        double avgSelectivity = 0;
        for (JobProfile profile : comparable) {
            // restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
            avgSelectivity += 1.0 * profile.getMapOutputBytes() / Math.max(profile.getTotalInputBytes(), profile.getTotalMapTasks());
        }
        return avgSelectivity / comparable.size();
    }

    private Long handleNoReduceHistory(JobProfile job, Double avgSelectivity) {
        if (avgSelectivity == 0 || job.getCompletedMaps() == 0) {
            // our selectivity or map rate data is unreliable
            // just return default duration
            logger.debug("No data to compute reduce for " + job.getName() + ". Using default");
            return conf.getLong(POSUMConfiguration.AVERAGE_TASK_DURATION,
                    POSUMConfiguration.AVERAGE_TASK_DURATION_DEFAULT);
        }

        // calculate the current map rate and assume reduce rate is the same

        // restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
        double mapRate = 1.0 * Math.max(job.getTotalInputBytes(), job.getTotalMapTasks()) / job.getAvgMapDuration();
        // we assume the reduce processing rate is the same as the map processing rate
        Double inputPerTask = Math.max(job.getTotalInputBytes(), 1) * avgSelectivity / job.getTotalReduceTasks();
        Double duration = inputPerTask / mapRate;
        logger.debug("Reduce duration computed based on map data for " + job.getId() + " as " + duration + "from mapRate=" + mapRate + " and selectivity=" + avgSelectivity);
        return duration.longValue();
    }

    private Long predictReduceTaskDuration(JobProfile job) {
        if (job.getAvgReduceDuration() != 0)
            return job.getAvgReduceDuration();

        // calculate average duration based on map selectivity and historical processing rates

        double avgSelectivity = calculateMapTaskSelectivity(job);

        List<JobProfile> comparable = getComparableProfiles(job, TaskType.REDUCE);
        if (comparable.size() < 1) {
            // non-existent or irrelevant historical data
            return handleNoReduceHistory(job, avgSelectivity);
        }
        boolean relevantHistory = comparable.get(0).getReducerClass().equals(job.getReducerClass());

        // we have reduce history; calculate reduce rate
        double avgReduceRate = 0, avgReduceDuration = 0;
        int comparableNo = 0;
        for (JobProfile profile : comparable) {
            if (profile.getTotalReduceTasks() < 1)
                continue;
            comparableNo++;
            if (avgSelectivity == 0 || !relevantHistory)
                // our selectivity or reduce rate data is unreliable
                // just compute average reduce duration of historical jobs
                avgReduceDuration += profile.getAvgReduceDuration();
            else
                // calculate reduce rate; restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
                avgReduceRate += 1.0 * Math.max(profile.getReduceInputBytes(), profile.getTotalReduceTasks()) /
                        profile.getAvgReduceDuration();
        }
        if (comparableNo < 1)
            // non-existent or irrelevant historical data
            return handleNoReduceHistory(job, avgSelectivity);
        if (avgSelectivity == 0 || !relevantHistory)
            // our selectivity or reduce rate data is unreliable
            // just compute average reduce duration of historical jobs<commit_msg>inputs per task take more care in including task numbers<commit_after>package org.apache.hadoop.tools.posum.simulator.predictor;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapreduce.v2.api.records.TaskType;
import org.apache.hadoop.tools.posum.common.records.dataentity.DataEntityType;
import org.apache.hadoop.tools.posum.common.records.dataentity.JobProfile;
import org.apache.hadoop.tools.posum.common.records.dataentity.TaskProfile;
import org.apache.hadoop.tools.posum.common.util.POSUMConfiguration;
import org.apache.hadoop.tools.posum.common.util.POSUMException;

import java.util.List;

/**
 * Created by ane on 2/9/16.
 */
public class StandardPredictor extends JobBehaviorPredictor {

    private static final Log logger = LogFactory.getLog(StandardPredictor.class);

    private List<JobProfile> getComparableProfiles(JobProfile job, TaskType type) {
        int bufferLimit = conf.getInt(POSUMConfiguration.PREDICTION_BUFFER,
                POSUMConfiguration.PREDICTION_BUFFER_DEFAULT);
        // get past jobs with the same name
        List<JobProfile> comparable = getDataStore().find(
                DataEntityType.JOB_HISTORY,
                type.equals(TaskType.MAP) ? "mapperClass" : "reducerClass",
                type.equals(TaskType.MAP) ? job.getMapperClass() : job.getReducerClass(),
                -bufferLimit,
                bufferLimit
        );
        if (comparable.size() < 1) {
            // get past jobs at least by the same user
            comparable = getDataStore().find(
                    DataEntityType.JOB_HISTORY,
                    "user",
                    job.getUser(),
                    -bufferLimit,
                    bufferLimit
            );
        }
        return comparable;
    }

    @Override
    public Long predictJobDuration(String jobId) {
        JobProfile job = getDataStore().findById(DataEntityType.JOB, jobId);
        return predictMapTaskDuration(job) * job.getTotalMapTasks() +
                predictReduceTaskDuration(job) * job.getTotalReduceTasks();
    }

    @Override
    public Long predictLocalMapTaskDuration(String jobId) {
        return predictTaskDuration(jobId, TaskType.MAP);
    }


    private Long predictMapTaskDuration(JobProfile job) {
        if (job.getAvgMapDuration() != 0)
            return job.getAvgMapDuration();
        // we have no information about this job; predict from history
        List<JobProfile> comparable = getComparableProfiles(job, TaskType.MAP);
        if (comparable.size() < 1) {
            logger.debug("No map history data for " + job.getId() + ". Using default");
            return conf.getLong(POSUMConfiguration.AVERAGE_TASK_DURATION,
                    POSUMConfiguration.AVERAGE_TASK_DURATION_DEFAULT);
        }
        double avgMapRate = 0;
        for (JobProfile profile : comparable) {
            //restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
            avgMapRate += 1.0 * Math.max(profile.getTotalInputBytes(), profile.getTotalMapTasks()) / profile.getAvgMapDuration();
        }
        avgMapRate /= comparable.size();
        // restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
        Double duration = Math.max(job.getTotalInputBytes(), job.getTotalMapTasks()) / avgMapRate;
        logger.debug("Map duration for " + job.getId() + " should be " + job.getTotalInputBytes() + " / " + avgMapRate + " = " + duration);
        return duration.longValue();
    }

    private double calculateMapTaskSelectivity(JobProfile job) {
        // we try to compute selectivity from the map history
        List<JobProfile> comparable = getComparableProfiles(job, TaskType.MAP);
        if (comparable.size() < 1 || !comparable.get(0).getMapperClass().equals(job.getMapperClass())) {
            // there is no history, or it is not relevant for selectivity
            if (job.getCompletedMaps() > 0) {
                // we know the current selectivity
                // restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
                Long parsedInputBytes = job.getCompletedMaps() * Math.max(job.getTotalInputBytes() / job.getTotalMapTasks(), 1);
                return 1.0 * job.getMapOutputBytes() / parsedInputBytes;
            }
            // we don't know anything about selectivity
            return 0;
        }
        double avgSelectivity = 0;
        for (JobProfile profile : comparable) {
            // restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
            avgSelectivity += 1.0 * profile.getMapOutputBytes() / Math.max(profile.getTotalInputBytes(), profile.getTotalMapTasks());
        }
        return avgSelectivity / comparable.size();
    }

    private Long handleNoReduceHistory(JobProfile job, Double avgSelectivity) {
        if (avgSelectivity == 0 || job.getCompletedMaps() == 0) {
            // our selectivity or map rate data is unreliable
            // just return default duration
            logger.debug("No data to compute reduce for " + job.getName() + ". Using default");
            return conf.getLong(POSUMConfiguration.AVERAGE_TASK_DURATION,
                    POSUMConfiguration.AVERAGE_TASK_DURATION_DEFAULT);
        }

        // calculate the current map rate and assume reduce rate is the same

        // restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
        Long parsedInputBytes = job.getCompletedMaps() * Math.max(job.getTotalInputBytes() / job.getTotalMapTasks(), 1);
        double mapRate = 1.0 * parsedInputBytes / job.getAvgMapDuration();
        // we assume the reduce processing rate is the same as the map processing rate
        Double inputPerTask = Math.max(job.getTotalInputBytes(), 1) * avgSelectivity / job.getTotalReduceTasks();
        Double duration = inputPerTask / mapRate;
        logger.debug("Reduce duration computed based on map data for " + job.getId() + " as " + duration + "from mapRate=" + mapRate + " and selectivity=" + avgSelectivity);
        return duration.longValue();
    }

    private Long predictReduceTaskDuration(JobProfile job) {
        if (job.getAvgReduceDuration() != 0)
            return job.getAvgReduceDuration();

        // calculate average duration based on map selectivity and historical processing rates

        double avgSelectivity = calculateMapTaskSelectivity(job);

        List<JobProfile> comparable = getComparableProfiles(job, TaskType.REDUCE);
        if (comparable.size() < 1) {
            // non-existent or irrelevant historical data
            return handleNoReduceHistory(job, avgSelectivity);
        }
        boolean relevantHistory = comparable.get(0).getReducerClass().equals(job.getReducerClass());

        // we have reduce history; calculate reduce rate
        double avgReduceRate = 0, avgReduceDuration = 0;
        int comparableNo = 0;
        for (JobProfile profile : comparable) {
            if (profile.getTotalReduceTasks() < 1)
                continue;
            comparableNo++;
            if (avgSelectivity == 0 || !relevantHistory)
                // our selectivity or reduce rate data is unreliable
                // just compute average reduce duration of historical jobs
                avgReduceDuration += profile.getAvgReduceDuration();
            else {
                // calculate reduce rate; restrict to a minimum of 1 byte per task to avoid multiplication or division by zero
                Double inputPerTask = 1.0 * Math.max(profile.getReduceInputBytes() / profile.getTotalReduceTasks(), 1);
                avgReduceRate += inputPerTask / profile.getAvgReduceDuration();
            }
        }
        if (comparableNo < 1)
            // non-existent or irrelevant historical data
            return handleNoReduceHistory(job, avgSelectivity);
        if (avgSelectivity == 0 || !relevantHistory)
            // our selectivity or reduce rate data is unreliable
            // just compute average reduce duration of historical jobs
{'fraction_non_alphanumeric': 0.08928125779885202, 'fraction_numerical': 0.005552782630396806, 'mean_word_length': 2.5112814895947424, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3878989', 'n_tokens_mistral': 4289, 'n_tokens_neox': 4077, 'n_words': 1147}