txtファイルからの読み取りとHBaseへの書き込み

Question

私はtxtファイルから読み込み、HBaseに書き込もうとしています。

Job Class 
    Job job = new Job(conf, "HWriterJob"); 
     job.setJarByClass(HWriterJob.class); 
     FileInputFormat.setInputPaths(job, new Path(otherArgs[0])); 

     job.setMapperClass(TokenizerMapper.class); 
     job.setOutputKeyClass(ImmutableBytesWritable.class); 
     job.setOutputValueClass(Put.class); 

     TableMapReduceUtil.initTableReducerJob(table,null,job); 

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

     String line = value.toString(); 

     StringTokenizer st = new StringTokenizer(line, "|"); 
     String result[] = new String[st.countTokens()]; 
     int i = 0; 
     while (st.hasMoreTokens()) { 
      result[i] = st.nextToken(); 
      i++; 
     } 

     Map<ImmutableBytesWritable,Put> resultSet = writeToHBase(result); 

     for (Map.Entry<ImmutableBytesWritable,Put> entry : resultSet.entrySet()) { 
      context.write(new Text(entry.getValue().getRow()), entry.getValue()); 
     } 
    } 

Reducer Class 

public void reduce(Text key, Iterable<Put> values, Context context) 
      throws IOException, InterruptedException { 

     for (Put val : values) { 
      context.write(key, val); 
     } 
    } 

しかし、私は同じことをするには大変です。

私は、次のようなエラーとjava.lang.ClassCastExceptionを取得しています：org.apache.hadoop.io.LongWritableはMapOutputKeyClassとしてLongWritableにorg.apache.hadoop.io.Textに

Answer 1

どうやらMRのデフォルト値をキャストすることはできませんあなたのケースではTextであり、エラーである必要があります。

はjob.setMapOutputKeyClass（Text.class）と もjob.setMapOutputValueClass適宜設定を設定してみてください。

Answer 2

  @Job Class 
      Job job = new Job(conf, "HWriterJob"); 
    job.setJarByClass(HWriterJob.class); 
    FileInputFormat.setInputPaths(job, new Path(otherArgs[0])); 

    job.setMapperClass(TokenizerMapper.class); 
    TextInputFormat.setInputPaths(job, new Path(args[0])); 
    job.setInputFormatClass(TextInputFormat.class); 
    job.setOutputKeyClass(ImmutableBytesWritable.class); 
     job.setOutputValueClass(Put.class); 
    job.setOutputFormatClass(TableOutputFormat.class); 

    job.setNumReduceTasks(0); 
    System.exit(job.waitForCompletion(true) ? 0 : 1); 
      @Mapper 
      Map<ImmutableBytesWritable,Put> resultSet = writeToHBase(result); 

    for (Map.Entry<ImmutableBytesWritable,Put> entry : resultSet.entrySet()) { 
     context.write(entry.getKey(), entry.getValue()); 
    }