Java 类org.apache.hadoop.hbase.mapreduce.Import 实例源码

private static boolean runImport(String[] args, Configuration configuration) throws IOException, InterruptedException, ClassNotFoundException {
    // need to make a copy of the configuration because to make sure different temp dirs are used.
    GenericOptionsParser opts = new GenericOptionsParser(new Configuration(configuration), args);
    Configuration newConf = opts.getConfiguration();
    args = opts.getRemainingArgs();
    Job job = Import.createSubmittableJob(newConf, args);
    job.waitForCompletion(false);
    return job.isSuccessful();
}

@Test
@Category(KnownGap.class)
public void testMapReduce() throws IOException, ClassNotFoundException, InterruptedException {
  Table oldTable = getConnection().getTable(TABLE_NAME);

  // Put a value.
  byte[] rowKey = dataHelper.randomData("testrow-");
  byte[] qual = dataHelper.randomData("testQualifier-");
  byte[] value = dataHelper.randomData("testValue-");
  Put put = new Put(rowKey);
  put.addColumn(COLUMN_FAMILY, qual, value);
  oldTable.put(put);

  // Assert the value is there.
  Get get = new Get(rowKey);
  Result result = oldTable.get(get);
  List<Cell> cells = result.listCells();
  Assert.assertEquals(1, cells.size());
  Assert.assertArrayEquals(CellUtil.cloneValue(cells.get(0)), value);

  // Run the export.
  Configuration conf = getConnection().getConfiguration();

  //conf.set("fs.defaultFS", "file:///");
  FileSystem dfs = IntegrationTests.getMiniCluster().getFileSystem();
  String tempDir = "hdfs://" + dfs.getCanonicalServiceName() + "/tmp/backup";

  String[] args = new String[]{
      TABLE_NAME.getNameAsString(),
      tempDir
  };
  Job job = Export.createSubmittableJob(conf, args);
  // So it looks for jars in the local FS, not HDFS.
  job.getConfiguration().set("fs.defaultFS", "file:///");
  Assert.assertTrue(job.waitForCompletion(true));

  // Create new table.
  TableName newTableName = IntegrationTests.newTestTableName();
  Table newTable = getConnection().getTable(newTableName);

  // Change for method in IntegrationTests
  Admin admin = getConnection().getAdmin();
  HColumnDescriptor hcd = new HColumnDescriptor(IntegrationTests.COLUMN_FAMILY);
  HTableDescriptor htd = new HTableDescriptor(newTableName);
  htd.addFamily(hcd);
  admin.createTable(htd);

  // Run the import.
  args = new String[]{
      newTableName.getNameAsString(),
      tempDir
  };
  job = Import.createSubmittableJob(conf, args);
  job.getConfiguration().set("fs.defaultFS", "file:///");
  Assert.assertTrue(job.waitForCompletion(true));

  // Assert the value is there.
  get = new Get(rowKey);
  result = newTable.get(get);
  cells = result.listCells();
  Assert.assertEquals(1, cells.size());
  Assert.assertArrayEquals(CellUtil.cloneValue(cells.get(0)), value);
}