private SimpleJob addBigJoinJob(SimpleJob job)
                throws IOException {
        Configuration conf = job.getConfiguration();
        String[] labels = conf.getStrings(SimpleJob.LABELS);
        String separator = conf.get(SimpleJob.SEPARATOR);
        boolean regex = conf.getBoolean(SimpleJob.SEPARATOR_REGEX, false);
        boolean formatIgnored = conf.getBoolean(SimpleJob.FORMAT_IGNORED, false);

        SimpleJob joinJob = new SimpleJob(conf, jobName, true);
        setConfiguration(joinJob, labels, separator, formatIgnored, regex);

        int type = conf.getInt(SimpleJob.READER_TYPE, -1);
        Configuration joinConf = joinJob.getConfiguration();
        joinConf.setInt(SimpleJob.READER_TYPE, type);
        joinConf.setStrings(SimpleJob.MASTER_LABELS, conf.getStrings(SimpleJob.MASTER_LABELS));
        if (type == SimpleJob.SINGLE_COLUMN_JOIN_READER) {
            joinConf.set(SimpleJob.JOIN_MASTER_COLUMN, conf.get(SimpleJob.JOIN_MASTER_COLUMN));
            joinConf.set(SimpleJob.JOIN_DATA_COLUMN, conf.get(SimpleJob.JOIN_DATA_COLUMN));
        } else if (type == SimpleJob.SOME_COLUMN_JOIN_READER) {
            joinConf.setStrings(SimpleJob.JOIN_MASTER_COLUMN, conf.getStrings(SimpleJob.JOIN_MASTER_COLUMN));
            joinConf.setStrings(SimpleJob.JOIN_DATA_COLUMN, conf.getStrings(SimpleJob.JOIN_DATA_COLUMN));
        }
        joinConf.set(SimpleJob.MASTER_PATH, conf.get(SimpleJob.MASTER_PATH));
        joinConf.set(SimpleJob.MASTER_SEPARATOR, conf.get(SimpleJob.MASTER_SEPARATOR));

        joinJob.setMapOutputKeyClass(Key.class);
        joinJob.setMapOutputValueClass(Value.class);

        joinJob.setPartitionerClass(SimplePartitioner.class);
        joinJob.setGroupingComparatorClass(SimpleGroupingComparator.class);
        joinJob.setSortComparatorClass(SimpleSortComparator.class);

        joinJob.setSummarizer(JoinSummarizer.class);

        if (!job.isMapper() && !job.isReducer()) {
            joinConf.setBoolean(SimpleJob.ONLY_JOIN, true);
            joinJob.setOutputKeyClass(Value.class);
            joinJob.setOutputValueClass(NullWritable.class);
        } else {
            joinJob.setOutputKeyClass(Key.class);
            joinJob.setOutputValueClass(Value.class);
        }

        return joinJob;
    }
private SimpleJob addBigJoinJob ( SimpleJob job ) throws IOException { Configuration conf = job . getConfiguration ( ) ; String [ ] labels = conf . getStrings ( SimpleJob . LABELS ) ; String separator = conf . get ( SimpleJob . SEPARATOR ) ; boolean regex = conf . getBoolean ( SimpleJob . SEPARATOR_REGEX , false ) ; boolean formatIgnored = conf . getBoolean ( SimpleJob . FORMAT_IGNORED , false ) ; SimpleJob joinJob = new SimpleJob ( conf , jobName , true ) ; setConfiguration ( joinJob , labels , separator , formatIgnored , regex ) ; int type = conf . getInt ( SimpleJob . READER_TYPE , - 1 ) ; Configuration joinConf = joinJob . getConfiguration ( ) ; joinConf . setInt ( SimpleJob . READER_TYPE , type ) ; joinConf . setStrings ( SimpleJob . MASTER_LABELS , conf . getStrings ( SimpleJob . MASTER_LABELS ) ) ; if ( type == SimpleJob . SINGLE_COLUMN_JOIN_READER ) { joinConf . set ( SimpleJob . JOIN_MASTER_COLUMN , conf . get ( SimpleJob . JOIN_MASTER_COLUMN ) ) ; joinConf . set ( SimpleJob . JOIN_DATA_COLUMN , conf . get ( SimpleJob . JOIN_DATA_COLUMN ) ) ; } else if ( type == SimpleJob . SOME_COLUMN_JOIN_READER ) { joinConf . setStrings ( SimpleJob . JOIN_MASTER_COLUMN , conf . getStrings ( SimpleJob . JOIN_MASTER_COLUMN ) ) ; joinConf . setStrings ( SimpleJob . JOIN_DATA_COLUMN , conf . getStrings ( SimpleJob . JOIN_DATA_COLUMN ) ) ; } joinConf . set ( SimpleJob . MASTER_PATH , conf . get ( SimpleJob . MASTER_PATH ) ) ; joinConf . set ( SimpleJob . MASTER_SEPARATOR , conf . get ( SimpleJob . MASTER_SEPARATOR ) ) ; joinJob . setMapOutputKeyClass ( Key . class ) ; joinJob . setMapOutputValueClass ( Value . class ) ; joinJob . setPartitionerClass ( SimplePartitioner . class ) ; joinJob . setGroupingComparatorClass ( SimpleGroupingComparator . class ) ; joinJob . setSortComparatorClass ( SimpleSortComparator . class ) ; joinJob . setSummarizer ( JoinSummarizer . class ) ; if ( ! job . isMapper ( ) && ! job . isReducer ( ) ) { joinConf . setBoolean ( SimpleJob . ONLY_JOIN , true ) ; joinJob . setOutputKeyClass ( Value . class ) ; joinJob . setOutputValueClass ( NullWritable . class ) ; } else { joinJob . setOutputKeyClass ( Key . class ) ; joinJob . setOutputValueClass ( Value . class ) ; } return joinJob ; }
create big join SimpleJob
@param job job that big join is set.
@return big join {@link SimpleJob}
@throws IOException