从Java客户端运行LoadIncrementAlhfiles



我想从我的java客户端代码调用 hbase org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles /user/myuser/map_data/hfiles mytable方法。

运行应用程序时,我会得到以下例外:

org.apache.hadoop.hbase.io.hfile.CorruptHFileException: Problem reading HFile Trailer from file webhdfs://myserver.de:50070/user/myuser/map_data/hfiles/b/b22db8e263b74a7dbd8e36f9ccf16508
    at org.apache.hadoop.hbase.io.hfile.HFile.pickReaderVersion(HFile.java:477)
    at org.apache.hadoop.hbase.io.hfile.HFile.createReader(HFile.java:520)
    at org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles.groupOrSplit(LoadIncrementalHFiles.java:632)
    at org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles$3.call(LoadIncrementalHFiles.java:549)
    at org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles$3.call(LoadIncrementalHFiles.java:546)
    at java.util.concurrent.FutureTask.run(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
Caused by: java.lang.RuntimeException: native snappy library not available: this version of libhadoop was built without snappy support.
    at org.apache.hadoop.io.compress.SnappyCodec.checkNativeCodeLoaded(SnappyCodec.java:65)
    at org.apache.hadoop.io.compress.SnappyCodec.getDecompressorType(SnappyCodec.java:193)
    at org.apache.hadoop.io.compress.CodecPool.getDecompressor(CodecPool.java:178)
    at org.apache.hadoop.hbase.io.compress.Compression$Algorithm.getDecompressor(Compression.java:327)
    at org.apache.hadoop.hbase.io.compress.Compression.decompress(Compression.java:422)
    at org.apache.hadoop.hbase.io.encoding.HFileBlockDefaultDecodingContext.prepareDecoding(HFileBlockDefaultDecodingContext.java:90)
    at org.apache.hadoop.hbase.io.hfile.HFileBlock.unpack(HFileBlock.java:529)
    at org.apache.hadoop.hbase.io.hfile.HFileBlock$AbstractFSReader$1.nextBlock(HFileBlock.java:1350)
    at org.apache.hadoop.hbase.io.hfile.HFileBlock$AbstractFSReader$1.nextBlockWithBlockType(HFileBlock.java:1356)
    at org.apache.hadoop.hbase.io.hfile.HFileReaderV2.<init>(HFileReaderV2.java:149)
    at org.apache.hadoop.hbase.io.hfile.HFileReaderV3.<init>(HFileReaderV3.java:77)
    at org.apache.hadoop.hbase.io.hfile.HFile.pickReaderVersion(HFile.java:467)
    ... 8 more

从我的hadoop服务器上的控制台运行上面的hbase ...命令完美工作。但是,当我尝试使用HBase/Hadoop客户端库从我的Java代码运行它们时,它会失败,但有例外!

在这里代码段:

public static void main(String[] args) {
    try {
        Configuration conf = loginFromKeyTab("REALM.DE", "server.de", "user", "C:/user.keytab");
        conf.set("fs.webhdfs.impl", org.apache.hadoop.hdfs.web.WebHdfsFileSystem.class.getName());
        conf.set("hbase.zookeeper.quorum", "server1.de,server2.de,server3.de");
        conf.set("zookeeper.znode.parent", "/hbase-secure");
        conf.set("hbase.master.kerberos.principal", "hbase/_HOST@REALM.DE");
        conf.set("hbase.regionserver.kerberos.principal", "hbase/_HOST@REALM.DE");
        conf.set("hbase.security.authentication", "kerberos");
        Connection connection = ConnectionFactory.createConnection(conf);
        Table table = connection.getTable(TableName.valueOf("mytable"));
        RegionLocator locator = connection.getRegionLocator(table.getName());
        Job job = Job.getInstance(conf, "Test Bulk Load"); 
        //HFileOutputFormat2.configureIncrementalLoad(job, table, locator);     
        //Configuration conf2 = job.getConfiguration();
        LoadIncrementalHFiles loader = new LoadIncrementalHFiles(conf);
        loader.doBulkLoad(new Path(HDFS_PATH), connection.getAdmin(), table, locator);
    } catch(Exception e) {
        e.printStackTrace();
    }
}

我需要为项目增加依赖性吗?但是如何/何处/哪个版本?

我正在与HDP 2.5合作,其中包含HBase 1.1.2和Hadoop 2.7.3

我为我的问题找到了另一个解决方案:我正在使用一个运行Process实例的Java程序,该程序自动调用LoadIncrementalHFiles方法(直接在Hadoop节点上运行),而不是使用LoadIncrementalHFiles类本身在我的代码中!

在这里,我的解决方案的代码段:

TreeSet<String> subDirs = getHFileDirectories(new Path(HDFS_OUTPUT_PATH), conf);        // The HDFS_OUTPUT_PATH directory contains many HFile sub-directories
for(String hFileDir : subDirs) {
    String pathToReadFrom = HDFS_OUTPUT_PATH + "/" + hFileDir;
    String[] execCode = {"hbase", "org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles", "-Dcreate.table=no", pathToReadFrom, "mytable"};       // Important: Separate each parameter here!!!
    ProcessBuilder pb = new ProcessBuilder(execCode);
    pb.redirectErrorStream(true);
    final Process p = pb.start();
    new Thread(new Runnable() {
        public void run() {
            BufferedReader input = new BufferedReader(new InputStreamReader(p.getInputStream()));
            String line = null; 
            try {
                while ((line = input.readLine()) != null)
                    System.out.println(line);
            } catch (IOException e) {
                    e.printStackTrace();
             }
        }
    }).start();
    p.waitFor();
    int exitCode = p.exitValue();
    System.out.println(" ==> Exit Code: " + exitCode);
}
System.out.println("Finished");

如果有人有另一个解决方案(例如,如何直接在代码中使用LoadIncrementalHFiles类),请告诉我。谢谢!

相关内容

  • 没有找到相关文章

最新更新