HBaseOutput implements RandomOutput, BulkOutput, ProvidesAlias, ProvidesValidations { @Override public void applyBulkMutations(List<Tuple2<MutationType, Dataset<Row>>> planned) { for (Tuple2<MutationType, Dataset<Row>> mutationDataset : planned) { BulkHBaseMutatorFunction mutatorFunction; switch (mutationDataset._1()) { case UPSERT: case DELETE: mutatorFunction = new BulkHBaseMutatorFunction(config, mutationDataset._1()); break; default: LOG.error("Unsupported bulk mutation type: {}", mutationDataset._1()); throw new RuntimeException("Unsupported bulk mutation type: " + mutationDataset._1()); } mutationDataset._2().foreachPartition(mutatorFunction); } }  @Override void configure(Config config); @Override Iterable<Row> getExistingForFilters(Iterable<Row> filters); @Override void applyRandomMutations(List<Row> plannedRows); @Override Set<MutationType> getSupportedRandomMutationTypes(); @Override Set<MutationType> getSupportedBulkMutationTypes(); @Override void applyBulkMutations(List<Tuple2<MutationType, Dataset<Row>>> planned); @Override String getAlias(); @Override Validations getValidations();  }
@Test public void testApplyBulkMutations() throws Exception { Table table = connection.getTable(TableName.valueOf(TABLE)); Config config = ConfigUtils.configFromResource("/hbase/hbase-output-test.conf").getConfig("output"); config = config.withValue("zookeeper", ConfigValueFactory.fromAnyRef("localhost:" + utility.getZkCluster().getClientPort())); HBaseOutput output = new HBaseOutput(); output.configure(config); Dataset<Row> upserts = createBulkMutations(INPUT_ROWS); Dataset<Row> deletes = createBulkMutations(INPUT_ROWS); List<Tuple2<MutationType, Dataset<Row>>> bulk1 = Lists.newArrayList(); bulk1.add(new Tuple2<>(MutationType.UPSERT, upserts)); List<Tuple2<MutationType, Dataset<Row>>> bulk2 = Lists.newArrayList(); bulk2.add(new Tuple2<>(MutationType.DELETE, deletes)); List<Tuple2<MutationType, Dataset<Row>>> bulk3 = Lists.newArrayList(); bulk3.add(new Tuple2<>(MutationType.UPSERT, upserts)); bulk3.add(new Tuple2<>(MutationType.DELETE, deletes)); output.applyBulkMutations(bulk1); scanAndCountTable(table, INPUT_ROWS * 4); output.applyBulkMutations(bulk2); scanAndCountTable(table, 0); output.applyBulkMutations(bulk3); scanAndCountTable(table, 0); }