HiveOutput implements BulkOutput, ProvidesAlias, ProvidesValidations { @Override public void applyBulkMutations(List<Tuple2<MutationType, Dataset<Row>>> planned) { for (Tuple2<MutationType, Dataset<Row>> plan : planned) { MutationType mutationType = plan._1(); Dataset<Row> mutation = (doesAlignColumns) ? alignColumns(plan._2()) : plan._2(); DataFrameWriter<Row> writer = mutation.write(); if (partitionColumns != null) { writer = writer.partitionBy(partitionColumns); } if (options != null) { writer = writer.options(options); } switch (mutationType) { case INSERT: writer = writer.mode(SaveMode.Append); break; case OVERWRITE: writer = writer.mode(SaveMode.Overwrite); break; default: throw new RuntimeException("Hive output does not support mutation type: " + mutationType); } writer.insertInto(tableName); } }  @Override void configure(Config config); @Override void applyBulkMutations(List<Tuple2<MutationType, Dataset<Row>>> planned); @Override Set<MutationType> getSupportedBulkMutationTypes(); Dataset<Row> alignColumns(Dataset<Row> input); @Override String getAlias(); @Override Validations getValidations(); final static String TABLE_CONFIG; final static String PARTITION_BY_CONFIG; final static String LOCATION_CONFIG; final static String OPTIONS_CONFIG; final static String ALIGN_COLUMNS_CONFIG; final static String SPARK_SQL_CASE_SENSITIVE_CONFIG; }
@Ignore ("Needs review of Hive temporary folder creation") @Test public void applyBulkMutations() throws Exception { Map<String, Object> paramMap = new HashMap<>(); paramMap.put(HiveOutput.TABLE_CONFIG, "default"); config = ConfigFactory.parseMap(paramMap); HiveOutput hiveOutput = new HiveOutput(); assertNoValidationFailures(hiveOutput, config); hiveOutput.configure(config); hiveOutput.applyBulkMutations(appendPlannerSetup()); }