[HADOOP] hbase mapreduce에 Delete 또는 Put 오류 전달

hbase mapreduce에 Delete 또는 Put 오류 전달

hbase에서 mapreduce를 실행하는 중에 오류가 발생했습니다.

java.io.IOException: Pass a Delete or a Put
    at org.apache.hadoop.hbase.mapreduce.TableOutputFormat$TableRecordWriter.write(TableOutputFormat.java:125)
    at org.apache.hadoop.hbase.mapreduce.TableOutputFormat$TableRecordWriter.write(TableOutputFormat.java:84)
    at org.apache.hadoop.mapred.MapTask$NewDirectOutputCollector.write(MapTask.java:639)
    at org.apache.hadoop.mapreduce.TaskInputOutputContext.write(TaskInputOutputContext.java:80)
    at HBaseImporter$InnerMap.map(HBaseImporter.java:61)
    at HBaseImporter$InnerMap.map(HBaseImporter.java:1)
    at org.apache.hadoop.mapreduce.Mapper.run(Mapper.java:144)
    at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:764)
    at org.apache.hadoop.mapred.MapTask.run(MapTask.java:370)
    at org.apache.hadoop.mapred.LocalJobRunner$Job.run(LocalJobRunner.java:212)
12/11/27 16:16:50 INFO mapred.JobClient:  map 0% reduce 0%
12/11/27 16:16:50 INFO mapred.JobClient: Job complete: job_local_0001
12/11/27 16:16:50 INFO mapred.JobClient: Counters: 0

암호:

public class HBaseImporter extends Configured implements Tool  {    
    public static class InnerMap extends
TableMapper<Text, IntWritable> {
    IntWritable one = new IntWritable();

    public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {
    String val = new String(value.getValue(Bytes.toBytes("cf"), Bytes.toBytes("line")));
    String[] words = val.toString().split(" ");
        try {
                for(String word:words)
            {
            context.write(new Text(word), one);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public static class MyTableReducer extends TableReducer<Text, IntWritable, ImmutableBytesWritable>  {

    public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int i = 0;
            for (IntWritable val : values) {
                i += val.get();
            }
            Put put = new Put(Bytes.toBytes(key.toString()));
            put.add(Bytes.toBytes("cf"), Bytes.toBytes("count"), Bytes.toBytes(i));

            context.write(null, put);
    }
}


public int run(String args[]) throws Exception
{
    //Configuration conf = getConf();
     Configuration conf = HBaseConfiguration.create();
        conf.addResource(new Path("/home/trg/hadoop-1.0.4/conf/core-site.xml"));
        conf.addResource(new Path("/home/trg/hadoop-1.0.4/conf/hdfs-site.xml"));


    Job job = new Job(conf,"SM LogAnalyzer MR");

    job.setJarByClass(HBaseImporter.class);
    //FileInputFormat.setInputPaths(job, new Path(args[1]));
    //FileOutputFormat.setOutputPath(job, new Path("outyy"));
     //job.setOutputFormatClass(TextOutputFormat.class);
      job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

    //job.setMapperClass(InnerMap.class);
    Scan scan = new Scan();
    scan.setCaching(500);        // 1 is the default in Scan, which will be bad for MapReduce jobs
    scan.setCacheBlocks(false);
    TableMapReduceUtil.initTableMapperJob(
            "wc_in",        // input table
            scan,               // Scan instance to control CF and attribute selection
            InnerMap.class,     // mapper class
            Text.class,         // mapper output key
            IntWritable.class,  // mapper output value
            job);

    TableMapReduceUtil.initTableReducerJob(
            "word_count",        // output table
            MyTableReducer.class,    // reducer class
            job);
        job.setNumReduceTasks(1);

    job.setNumReduceTasks(0);

    return job.waitForCompletion(true)?0:1;
}

public static void main(String[] args) throws Exception {
    //Configuration conf = new HBaseConfiguration();
    //Job job = configureJob(conf, args);
    //System.exit(job.waitForCompletion(true) ? 0 : 1);

    String[] inArgs = new String[4]; 
    inArgs[0] = "HBaseImporter";
        inArgs[1] = "/user/trg/wc_in"; 
        inArgs[2] = "AppLogMRImport"; 
        inArgs[3] = "MessageDB"; 
        int res = ToolRunner.run(new Configuration(), new HBaseImporter(), inArgs);
        //int res = ToolRunner.run(new Configuration(), new HBaseImporter(), args);

    }
}

맵 출력 값 클래스를 IntWritable.class로 설정하고 있지만 PutOpt 개체를 예상하는 매퍼에서 호출되는 경우에도 여전히 TableOutputFormat.write입니다.

해결법

1. ==============================

   1.내 자신의 질문에 대답했다. 실수로 감속기 작업을 '0'으로 설정했습니다.

   내 자신의 질문에 대답했다. 실수로 감속기 작업을 '0'으로 설정했습니다.

    job.setNumReduceTasks(0);
   

   그래서 매퍼는 Put 객체가 Hbase 테이블에 직접 쓰는 것을 기대합니다. 위 라인을 수정하면이 문제가 해결됩니다.

2. ==============================

   2.문제는 다음 코드로 해결되지 않았습니다.

   문제는 다음 코드로 해결되지 않았습니다.

   job.setNumReduceTasks(0);
   

   그것은 오직 감소 단계를 비활성화하고 감소 단계에서 실제 문제를 건너 뛰는 데 도움이됩니다.

   게다가 구성에 문제가 있으므로 mapred-site.xml과 hbase-site.xml을 모두 Configuration 자원으로 포함해야합니다.

from https://stackoverflow.com/questions/13578963/pass-a-delete-or-a-put-error-in-hbase-mapreduce by cc-by-sa and MIT license