0

HBase テーブルを入力として取り、HDFS ファイルにダンプする MR ジョブを作成しています。複数のデータ ソースを使用する予定なので、MultipleInputs クラス (Hadoop から) を使用します。私は非常に単純な MR プログラムを作成しました (以下のソース コードを参照)。残念ながら、次のエラーが発生します。

java.lang.ClassCastException: org.apache.hadoop.io.LongWritable は org.apache.hadoop.hbase.io.ImmutableBytesWritable にキャストできません

疑似分散 Hadoop (1.2.0) と疑似分散 HBase (0.95.1-hadoop1) で実行しています。

完全なソース コードは次のとおりです。興味深いのは、multipleinputs の行 "MultipleInputs.addInputPath(job, inputPath1, TextInputFormat.class, TableMap.class);" をコメント アウトすると、MR ジョブが正常に実行されることです。

public class MixMR {

public static class TableMap extends TableMapper<Text, Text>  {
    public static final byte[] CF = "cf".getBytes();
    public static final byte[] ATTR1 = "c1".getBytes();

    public void map(ImmutableBytesWritable row, Result value, Context context) throws IOException, InterruptedException {

        String key = Bytes.toString(row.get());
        String val = new String(value.getValue(CF, ATTR1));

        context.write(new Text(key), new Text(val));
    }
}


public static class Reduce extends Reducer  <Object, Text, Object, Text> {
    public void reduce(Object key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {
        String ks = key.toString();
        for (Text val : values){
            context.write(new Text(ks), val);
        }

    }
}

public static void main(String[] args) throws Exception {
    Path inputPath1 = new Path(args[0]);
    Path outputPath = new Path(args[1]);

    String tableName1 = "test";

    Configuration config = HBaseConfiguration.create();
    Job job = new Job(config, "ExampleRead");
    job.setJarByClass(MixMR.class);     // class that contains mapper


    Scan scan = new Scan();
    scan.setCaching(500);        // 1 is the default in Scan, which will be bad for MapReduce jobs
    scan.setCacheBlocks(false);  // don't set to true for MR jobs
    scan.addFamily(Bytes.toBytes("cf"));

    TableMapReduceUtil.initTableMapperJob(
            tableName1,        // input HBase table name
              scan,             // Scan instance to control CF and attribute selection
              TableMap.class,   // mapper
              Text.class,             // mapper output key
              Text.class,             // mapper output value
              job);
    job.setReducerClass(Reduce.class);    // reducer class
    job.setOutputFormatClass(TextOutputFormat.class);  

    // inputPath1 here has no effect for HBase table
    MultipleInputs.addInputPath(job, inputPath1, TextInputFormat.class, TableMap.class);

    FileOutputFormat.setOutputPath(job, outputPath);

    job.waitForCompletion(true);
}

}

4

1 に答える 1

0

私は答えを得ました：次のステートメントで：TextInputFormat.classをTableInputFormat.classに置き換えます

MultipleInputs.addInputPath(ジョブ、inputPath1、TextInputFormat.class、TableMap.class);

于 2013-07-12T15:50:49.993 に答える