从MapReduce写入到Hive(初始化HCatOutputFormat)

Question

我写了一个MR脚本，它应该从HBase中加载数据并将它们转储到Hive中。连接到HBase是可以的，但是当我试图将数据保存到HIVE表中时，我得到以下错误消息:

 Failing Oozie Launcher, Main class [org.apache.oozie.action.hadoop.JavaMain], main() threw exception, org.apache.hive.hcatalog.common.HCatException : 2004 : HCatOutputFormat not initialized, setOutput has to be called
  org.apache.oozie.action.hadoop.JavaMainException: org.apache.hive.hcatalog.common.HCatException : 2004 : HCatOutputFormat not initialized, setOutput has to be called
  at org.apache.oozie.action.hadoop.JavaMain.run(JavaMain.java:58)
  at org.apache.oozie.action.hadoop.LauncherMain.run(LauncherMain.java:38)
  at org.apache.oozie.action.hadoop.JavaMain.main(JavaMain.java:36)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:606)
  at org.apache.oozie.action.hadoop.LauncherMapper.map(LauncherMapper.java:226)
  at org.apache.hadoop.mapred.MapRunner.run(MapRunner.java:54)
  at org.apache.hadoop.mapred.MapTask.runOldMapper(MapTask.java:430)
  at org.apache.hadoop.mapred.MapTask.run(MapTask.java:342)
  at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:168)
  at java.security.AccessController.doPrivileged(Native Method)
  at javax.security.auth.Subject.doAs(Subject.java:415)
  at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1548)
  at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:163)
  Caused by: org.apache.hive.hcatalog.common.HCatException : 2004 : HCatOutputFormat not initialized, setOutput has to be called
  at org.apache.hive.hcatalog.mapreduce.HCatBaseOutputFormat.getJobInfo(HCatBaseOutputFormat.java:118)
  at org.apache.hive.hcatalog.mapreduce.HCatBaseOutputFormat.getTableSchema(HCatBaseOutputFormat.java:61)
  at com.nrholding.t0_mr.main.DumpProductViewsAggHive.run(DumpProductViewsAggHive.java:254)
  at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:70)
  at com.nrholding.t0_mr.main.DumpProductViewsAggHive.main(DumpProductViewsAggHive.java:268)
  at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
  at java.lang.reflect.Method.invoke(Method.java:606)
  at org.apache.oozie.action.hadoop.JavaMain.run(JavaMain.java:55)
  ... 15 more

我正在检查:

表存在
• setOutput方法在getTableSchema和setSchema
之前被调用

下面是我的run方法:

@Override
public int run(String[] args) throws Exception {
    // Create configuration
    Configuration conf = this.getConf();
    String databaseName = null;
    String tableName = "test";
    // Parse arguments
    String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();
    getParams(otherArgs);
    // It is better to specify zookeeper quorum in CLI parameter -D hbase.zookeeper.quorum=zookeeper servers
    conf.set( "hbase.zookeeper.quorum",
    "cz-dc1-s-132.mall.local,cz-dc1-s-133.mall.local,"
    + "cz-dc1-s-134.mall.local,cz-dc1-s-135.mall.local,"
    + "cz-dc1-s-136.mall.local");
    // Create job
    Job job = Job.getInstance(conf, NAME);
    job.setJarByClass(DumpProductViewsAggHive.class);

    // Setup MapReduce job
    job.setReducerClass(Reducer.class);
    //job.setNumReduceTasks(0); // If reducer is not needed
    // Specify key / value
    job.setOutputKeyClass(Writable.class);
    job.setOutputValueClass(DefaultHCatRecord.class);
    // Input
    getInput(null, dateFrom, dateTo, job, caching, table);
    // Output
    // Ignore the key for the reducer output; emitting an HCatalog record as value
    job.setOutputFormatClass(HCatOutputFormat.class);
    HCatOutputFormat.setOutput(job, OutputJobInfo.create(databaseName, tableName, null));
    HCatSchema s = HCatOutputFormat.getTableSchema(conf);
    System.err.println("INFO: output schema explicitly set for writing:" + s);
    HCatOutputFormat.setSchema(job, s);
    // Execute job and return status
    return job.waitForCompletion(true) ? 0 : 1;
}

你知道怎么帮我吗?谢谢你！

Answer 1

使用说明:

HCatSchema s = HCatOutputFormat.getTableSchema(job.getConfiguration());

Answer 2

好的，我使用折旧法:

HCatSchema s = HCatOutputFormat.getTableSchema(job);

代替:

HCatSchema s = HCatOutputFormat.getTableSchema(conf);

它似乎可以工作