0
私はphoenixを使用してhbaseテーブルから読み込むmap reduceジョブを持っています。私はこの仕事の出力がHDFSにあり、次に別のマップ還元ジョブにフィードして、HBASEテーブルに更新することを望みます。これは私が試みたものです。私はこれを実行するとHadoop with phoenix：phoenixテーブルオブジェクトをhdfsファイルシステムに書き込む方法
public class Job1Driver extends Configured implements Tool {
@Override
public int run(String[] args) throws Exception {
final org.apache.hadoop.conf.Configuration jobConfiguration = super.getConf();
final Job job1 = Job.getInstance(jobConfiguration, jobConfiguration.get("mapreduce.job.name"));
final String selectQuery = "SELECT * FROM TABLE1 WHERE IS_SUMMARY_RECORD=false";
job1.setJarByClass(Job1Driver.class);
PhoenixMapReduceUtil.setInputCluster(job1, jobConfiguration.get("HBASE_URL"));
PhoenixMapReduceUtil.setInput(job1, Table1Writable.class, "TABLE1", selectQuery);
if (jobConfiguration.get("IS_FROZEN_DATA_AVAILABLE").equals("True")) {
MultipleInputs.addInputPath(job1,new Path(args[0]),
TextInputFormat.class, FrozenMapper.class);
}
MultipleInputs.addInputPath(job1,new Path(args[1]),
PhoenixInputFormat.class,ActiveMapper.class);
FileOutputFormat.setOutputPath(job1, new Path(args[2]));
job1.setMapOutputKeyClass(Text.class);
job1.setMapOutputValueClass(Table1Writable.class);
job1.setOutputKeyClass(NullWritable.class);
job1.setOutputValueClass(Table1Writable.class);
job1.setReducerClass(Job1Reducer.class);
boolean st = job1.waitForCompletion(true);
return st ? 0 : 1;
}
public static void main(String[] args) throws Exception {
Configuration conf = HBaseConfiguration.create();
int exitCode = ToolRunner.run(conf, new Job1Driver(), args);
System.exit(exitCode);
}
、私は書き込み可能な実装を使用して出力ディレクトリ
[email protected]
にこのような何かを取得しています、私はマッパーアウトからHDFSに書き込むことができますが、同じことは、減速のために働いていませんでる。私が逃した明白なことは何ですか？