java - MAPREDUCE 错误 : method write in interface TaskInputOutputContext 无法应用于给定类型

Question

package br.edu.ufam.anibrata;

import java.io.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;
import java.util.Arrays;
import java.util.HashSet;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MapFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;
import org.kohsuke.args4j.CmdLineException;
import org.kohsuke.args4j.CmdLineParser;
import org.kohsuke.args4j.Option;
import org.kohsuke.args4j.ParserProperties;

import tl.lin.data.array.ArrayListWritable;
import tl.lin.data.pair.PairOfStringInt;
import tl.lin.data.pair.PairOfWritables;
import br.edu.ufam.data.Dataset;

import com.google.gson.JsonSyntaxException;

public class BuildIndexWebTables extends Configured implements Tool {

    private static final Logger LOG = Logger.getLogger(BuildIndexWebTables.class);

    public static void main(String[] args) throws Exception 
    {
        ToolRunner.run(new BuildIndexWebTables(), args);
    }

    @Override
    public int run(String[] argv) throws Exception {
        // Creates a new job configuration for this Hadoop job.
        Args args = new Args();
        CmdLineParser parser = new CmdLineParser(args, ParserProperties.defaults().withUsageWidth(100));

        try 
        {
            parser.parseArgument(argv);
        }
        catch (CmdLineException e) 
        {
            System.err.println(e.getMessage());
            parser.printUsage(System.err);
            return -1;
        }

        Configuration conf = getConf();

        conf.setBoolean("mapreduce.map.output.compress", true);
        conf.setBoolean("mapreduce.map.output.compress", true);
        conf.set("mapreduce.map.failures.maxpercent", "10");
        conf.set("mapreduce.max.map.failures.percent", "10");
        conf.set("mapred.max.map.failures.percent", "10");
        conf.set("mapred.map.failures.maxpercent", "10");

        conf.setBoolean("mapred.compress.map.output", true);
        conf.set("mapred.map.output.compression.codec", "org.apache.hadoop.io.compress.SnappyCodec");
        conf.setBoolean("mapreduce.map.output.compress", true);

        /*String inputPrefixes = args[0];
        String outputFile = args[1];*/

        Job job = Job.getInstance(conf);

        /*FileInputFormat.addInputPath(job, new Path(inputPrefixes));
        FileOutputFormat.setOutputPath(job, new Path(outputFile));*/

        FileInputFormat.setInputPaths(job, new Path(args.input));
        FileOutputFormat.setOutputPath(job, new Path(args.output));

        FileOutputFormat.setCompressOutput(job, true);
        FileOutputFormat.setOutputCompressorClass(job,org.apache.hadoop.io.compress.GzipCodec.class);

        job.setMapperClass(BuildIndexWebTablesMapper.class);
        job.setReducerClass(BuildIndexWebTablesReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);     
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(PairOfWritables.class);
        //job.setOutputFormatClass(MapFileOutputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        /*job.setOutputFormatClass(TextOutputFormat.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);*/


        job.setJarByClass(BuildIndexWebTables.class);
        job.setNumReduceTasks(args.numReducers);
        //job.setNumReduceTasks(500);

        FileInputFormat.setInputPaths(job, new Path(args.input));
        FileOutputFormat.setOutputPath(job, new Path(args.output));

        System.out.println(Arrays.deepToString(FileInputFormat.getInputPaths(job)));

        // Delete the output directory if it exists already.
        Path outputDir = new Path(args.output);
        FileSystem.get(getConf()).delete(outputDir, true);

        long startTime = System.currentTimeMillis();
        job.waitForCompletion(true);
        System.out.println("Job Finished in " + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");

        return 0;
    }

    private BuildIndexWebTables() {}

    public static class Args 
    {
        @Option(name = "-input", metaVar = "[path]", required = true, usage = "input path")
        public String input;

        @Option(name = "-output", metaVar = "[path]", required = true, usage = "output path")
        public String output;

         @Option(name = "-reducers", metaVar = "[num]", required = false, usage = "number of reducers")
         public int numReducers = 1;
    }

    public static class BuildIndexWebTablesMapper extends Mapper<LongWritable, Text, Text, Text> {
        //public static final Log log = LogFactory.getLog(BuildIndexWebTablesMapper.class);
        private static final Text WORD = new Text();
        private static final Text OPVAL = new Text();

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

            // Log to stdout file
            System.out.println("Map key : TEST");

            //log to the syslog file
            //log.info("Map key "+ key);

            /*if(log.isDebugEanbled()){
                log.debug("Map key "+ key);
            }*/

            Dataset ds;
            String pgTitle; // Table page title

            List<String> tokens = new ArrayList<String>(); // terms for frequency and other data

            ds = Dataset.fromJson(value.toString());    // Get all text values from the json corpus
            String[][] rel = ds.getRelation();          // Extract relation from the first json 
            int numCols = rel.length;                   // Number of columns in the relation
            String[] attributes = new String[numCols];  // To store attributes for the relation

            for (int j = 0; j < numCols; j++) {         // Attributes of the relation
                attributes[j] = rel[j][0];
            }

            int numRows = rel[0].length;                //Number of rows of the relation
            //dsTabNum = ds.getTableNum();              // Gets the table number from json

            // Reads terms from relation and stores in tokens
            for (int i = 0; i < numRows; i++ ){
                for (int j = 0; j < numCols; j++ ){
                    String w = rel[i][j].toLowerCase().replaceAll("(^[^a-z]+|[^a-z]+$)", "");
                    if (w.length() == 0) 
                        continue;
                    else {
                        w = w + "|" + pgTitle + "." + j + "|" + i; // Concatenate the term/PageTitle.Column number/row number in term   
                        tokens.add(w);
                    }
                }
            }

            // Emit postings.
            for (String token : tokens){
                String[] tokenPart = token.split("|", -2);      // Split based on "|", -2(any negative) to split multiple times.
                String newkey = tokenPart[0] + "|" + tokenPart[1];
                WORD.set(newkey);   // Emit term as key
                //String valstr = Arrays.toString(Arrays.copyOfRange(tokenPart, 2, tokenPart.length)); // Emit rest of the string as value 
                String valstr = tokenPart[2];
                OPVAL.set(valstr);
                context.write(WORD,OPVAL);
            }
        }

    }

    public static class BuildIndexWebTablesReducer extends Reducer<Text, Text, Text, Text> {

        private static final Text TERM = new Text();
        private static final IntWritable TF = new IntWritable();
        private String PrevTerm = null;
        private int termFrequency = 0;

        @Override
        protected void reduce(Text key, Iterable<Text> textval, Context context) throws IOException, InterruptedException {

            Iterator<Text> iter = textval.iterator();
            IntWritable tnum = new IntWritable();
            ArrayListWritable<IntWritable> postings = new ArrayListWritable<IntWritable>();
            PairOfStringInt relColInfo = new PairOfStringInt();
            PairOfWritables keyVal = new PairOfWritables<PairOfStringInt, ArrayListWritable<IntWritable>>();

            if((!key.toString().equals(PrevTerm)) && (PrevTerm != null)) {
                String[] parseKey = PrevTerm.split("|", -2);
                TERM.set(parseKey[0]);
                relColInfo.set(parseKey[1],termFrequency);
                keyVal.set(relColInfo, postings);
                context.write(TERM, keyVal);
                termFrequency = 0;
                postings.clear();
            }

            PrevTerm = key.toString();

            while (iter.hasNext()) {
                int tupleset = Integer.parseInt(iter.next().toString());
                tnum.set(tupleset);
                postings.add(tnum);
                termFrequency++;        
            }
        }
    }
}`

我在编译时遇到了下面提到的错误。

[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:2.3.2:compile (default-compile) on project projeto-final: Compilation failure [ERROR] /home/cloudera/topicosBD-pis/topicosBD-pis/projeto-final/src/main/java/br/edu/ufam/anibrata/BuildIndexWebTables.java:[278,11] error: method write in interface TaskInputOutputContext cannot be applied to given types;

恰好发生这种情况的行是“context.write(TERM, keyVal);”。不过，这段代码有一些依赖于我的本地机器。我被这个错误困住了，因为我对它一无所知。如果有人可以帮助我了解问题的根源以及如何解决这个问题。我对 hadoop/mapreduce 很陌生。我尝试在 job.setOutputFormatClass(MapFileOutputFormat.class) 之间切换 OutputFormatClass；和job.setOutputFormatClass(TextOutputFormat.class);，它们都抛出相同的错误。我正在使用“mvn clean package”进行编译。

非常感谢任何帮助。

提前致谢。

Answer 1

如我所见，您正尝试在上下文中编写一个 Text 类型的键 (TERM) 和一个 PairOfWritables 类型的值 (keyval)，但是您的 reducer类使用 TEXT 类型的 VALUEOUT(最后一个)扩展 Reducer。您应该将 VALUEOUT 更改为正确的类型。

在你的情况下:

public static class BuildIndexWebTablesReducer extends Reducer<Text, Text, Text, PairOfWritables>