hdfs plain text write to hbase,Output directory not set
<p>In Map I read Hdfs file update to Hbase,</p>

<p>Version：hadoop 2.5.1 hbase 1.0.0</p>

<p>Exception as follows :</p>

<pre><code>Exception in thread "main" org.apache.hadoop.mapred.InvalidJobConfException: Output directory not set.
</code></pre>

<p>maybe there is something wrong with  </p>

<pre><code>job.setOutputFormatClass(TableOutputFormat.class); 
</code></pre>

<p>this line prompt：</p>

<pre><code>The method setOutputFormatClass(Class&lt;? extends OutputFormat&gt;) in the type Job is not applicable for the arguments (Class&lt;TableOutputFormat&gt;)
</code></pre>

<p>codes as follows:</p>

<pre><code>public class HdfsAppend2HbaseUtil extends Configured implements Tool{

    public static class HdfsAdd2HbaseMapper extends Mapper&lt;Text, Text, ImmutableBytesWritable, Put&gt;{

        public void map(Text ikey, Text ivalue, Context context) 
                throws IOException, InterruptedException {

            String oldIdList = HBaseHelper.getValueByKey(ikey.toString());

            StringBuffer sb = new StringBuffer(oldIdList);
            String newIdList = ivalue.toString();
            sb.append("\t" + newIdList);

            Put p = new Put(ikey.toString().getBytes());
            p.addColumn("idFam".getBytes(), "idsList".getBytes(), sb.toString().getBytes());
            context.write(new ImmutableBytesWritable(), p);

        }

    }

    public int run(String[] paths) throws Exception {

        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master,salve1");
        conf.set("hbase.zookeeper.property.clientPort", "2181");

        Job job = Job.getInstance(conf,"AppendToHbase");
        job.setJarByClass(cn.edu.hadoop.util.HdfsAppend2HbaseUtil.class);

        job.setInputFormatClass(KeyValueTextInputFormat.class);

        job.setMapperClass(HdfsAdd2HbaseMapper.class);
        job.setNumReduceTasks(0);

        job.setOutputFormatClass(TableOutputFormat.class); 

        job.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "reachableTable");

        FileInputFormat.setInputPaths(job, new Path(paths[0]));


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


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

    public static void main(String[] args) throws Exception {

        System.out.println("Append Start: ");

        long time1 = System.currentTimeMillis();
        long time2;
        String[] pathsStr = {Const.TwoDegreeReachableOutputPathDetail};

        int exitCode = ToolRunner.run(new HdfsAppend2HbaseUtil(), pathsStr);
        time2 = System.currentTimeMillis();
        System.out.println("Append Cost " + "\t" + (time2-time1)/1000 +" s");

        System.exit(exitCode);
    }
}
</code></pre>