ExcludeDeriver implements Deriver, ProvidesAlias, ProvidesValidations { @Override public Dataset<Row> derive(Map<String, Dataset<Row>> dependencies) throws Exception { Dataset<Row> compare, with; if (!dependencies.containsKey(compareDataset)) { throw new RuntimeException("Designated comparison target dataset is not a dependency: " + compareDataset); } else { compare = dependencies.get(compareDataset); } if (!dependencies.containsKey(withDataset)) { throw new RuntimeException("Designated comparison reference dataset is not a dependency: " + withDataset); } else { with = dependencies.get(withDataset); } return compare.join(with, JavaConversions.asScalaBuffer(fields).toList(), "leftanti"); }  @Override void configure(Config config); @Override Dataset<Row> derive(Map<String, Dataset<Row>> dependencies); @Override String getAlias(); @Override Validations getValidations(); static final String EXCLUSION_COMPARE_CONFIG; static final String EXCLUSION_WITH_CONFIG; static final String EXCLUSION_FIELDS_CONFIG; }
@Test public void derive() throws Exception { StructType existingSchema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.IntegerType, true), DataTypes.createStructField("field2", DataTypes.StringType, true)) ); List<Row> existingRows = Lists.newArrayList(); existingRows.add(RowFactory.create(1000, "Envelopes")); existingRows.add(RowFactory.create(1001, "Stamps")); Dataset<Row> existingDF = Contexts.getSparkSession().createDataFrame(existingRows, existingSchema); StructType newSchema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("field1", DataTypes.IntegerType, true), DataTypes.createStructField("field2", DataTypes.StringType, true), DataTypes.createStructField("field3", DataTypes.StringType, true) ) ); List<Row> newRows= Lists.newArrayList(); newRows.add(RowFactory.create(1000, "Envelopes", "Nope")); newRows.add(RowFactory.create(1001, "Stamps", "Nope")); newRows.add(RowFactory.create(1000, "Stamps", "Yep")); newRows.add(RowFactory.create(1000, "Staplers", "Yep")); newRows.add(RowFactory.create(1001, "Envelopes", "Yep")); Dataset<Row> newDF = Contexts.getSparkSession().createDataFrame(newRows, newSchema); Map<String, Dataset<Row>> dependencies = new HashMap<>(); dependencies.put("New", newDF); dependencies.put("Existing", existingDF); Map<String, Object> paramMap = new HashMap<>(); paramMap.put(ExcludeDeriver.EXCLUSION_COMPARE_CONFIG, "New"); paramMap.put(ExcludeDeriver.EXCLUSION_WITH_CONFIG, "Existing"); paramMap.put(ExcludeDeriver.EXCLUSION_FIELDS_CONFIG, Lists.newArrayList("field1", "field2")); config = ConfigFactory.parseMap(paramMap); ExcludeDeriver excludeDeriver = new ExcludeDeriver(); assertNoValidationFailures(excludeDeriver, config); excludeDeriver.configure(config); Dataset<Row> results = excludeDeriver.derive(dependencies); assertNotNull("Results is null", results); assertNotNull("Invalid schema", results.schema()); assertEquals("Invalid schema field count", 3, results.schema().fieldNames().length); assertEquals("Invalid schema field name", "field3", results.schema().fieldNames()[2]); assertEquals("Invalid schema field name", DataTypes.StringType, results.schema().fields()[2].dataType()); assertEquals("Invalid row count", 3, results.count()); }