Как мне создать собственный класс mapper в mapreduce

Question

У меня есть уникальное требование, когда мне нужно передать команду оболочки zip из текстового файла, и mapper обработает скрипт, который будет создавать файлы zip параллельно, используя только mapper.Я думаю, чтобы выполнить команду оболочки, используя exec в Java.Я немного застрял на том, как реализовать пользовательский маппер, поскольку мой вывод будет сжатым форматом.

Ниже приведен мой класс картографа -

import java.io.IOException;
import java.util.StringTokenizer;    
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class Map extends Mapper<LongWritable, Text, Text, NullWritable>{    

    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{

        String line= value.toString();
        StringTokenizer tokenizer= new StringTokenizer(line);

        while(tokenizer.hasMoreTokens()){
            value.set(tokenizer.nextToken());
            context.write(value,NullWritable.get());        
        }


    }

}

Класс процессора -

import org.apache.hadoop.util.ToolRunner;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;

public class ZipProcessor extends Configured implements Tool {

    public static void main(String [] args) throws Exception{
        int exitCode = ToolRunner.run(new ZipProcessor(), args);
        System.exit(exitCode);
    }

    public int run(String[] args) throws Exception {

        if(args.length!=2){

             System.err.printf("Usage: %s needs two arguments, input and output files\n", getClass().getSimpleName());
             return -1;
        }

        Configuration conf=new Configuration();

        Job job = Job.getInstance(conf,"zipping");
        job.setJarByClass(ZipProcessor.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputFormatClass(TextOutputFormat.class);    
        job.setMapperClass(Map.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));


        int returnValue = job.waitForCompletion(true) ? 0:1;

        if(job.isSuccessful()) {
            System.out.println("Job was successful");
        } else if(!job.isSuccessful()) {
            System.out.println("Job was not successful");           
        }

        return returnValue;


    }

}

Пример mapr.txt

zip -r "/folder1/file.zip" "sourceFolder"
zip -r "/folder2/file.zip" "sourceFolder"
zip -r "/folder3/file.zip" "sourceFolder"