PassthroughDeriver implements Deriver, ProvidesAlias { @Override public Dataset<Row> derive(Map<String, Dataset<Row>> dependencies) throws Exception { if (dependencies.isEmpty()) { throw new RuntimeException("Passthrough deriver requires at least one dependency"); } Iterator<Dataset<Row>> dependencyIterator = dependencies.values().iterator(); Dataset<Row> unioned = dependencyIterator.next(); while (dependencyIterator.hasNext()) { Dataset<Row> next = dependencyIterator.next(); if (!unioned.schema().equals(next.schema())) { throw new RuntimeException("All dependencies of the passthrough deriver must have the same schema"); } unioned = unioned.union(next); } return unioned; }  @Override void configure(Config config); @Override Dataset<Row> derive(Map<String, Dataset<Row>> dependencies); @Override String getAlias();  }
@Test public void testPassthrough() throws Exception { StructType schema = DataTypes.createStructType(Lists.<StructField>newArrayList( DataTypes.createStructField("col1", DataTypes.StringType, false))); Dataset<Row> dep1 = Contexts.getSparkSession().createDataFrame( Lists.newArrayList(RowFactory.create("a")), schema); Dataset<Row> dep2= Contexts.getSparkSession().createDataFrame( Lists.newArrayList(RowFactory.create("b")), schema); Map<String, Dataset<Row>> dependencies = Maps.newHashMap(); dependencies.put("dep1", dep1); dependencies.put("dep2", dep2); Deriver deriver = new PassthroughDeriver(); List<Row> result = deriver.derive(dependencies).collectAsList(); assertTrue(result.contains(RowFactory.create("a"))); assertTrue(result.contains(RowFactory.create("b"))); assertEquals(2, result.size()); }
@Test (expected = RuntimeException.class) public void testDifferentSchemas() throws Exception { StructType schema1 = DataTypes.createStructType(Lists.<StructField>newArrayList( DataTypes.createStructField("col1", DataTypes.StringType, false))); StructType schema2 = DataTypes.createStructType(Lists.<StructField>newArrayList( DataTypes.createStructField("col2", DataTypes.StringType, false))); Dataset<Row> dep1 = Contexts.getSparkSession().createDataFrame( Lists.newArrayList(RowFactory.create("a")), schema1); Dataset<Row> dep2= Contexts.getSparkSession().createDataFrame( Lists.newArrayList(RowFactory.create("b")), schema2); Map<String, Dataset<Row>> dependencies = Maps.newHashMap(); dependencies.put("dep1", dep1); dependencies.put("dep2", dep2); Deriver deriver = new PassthroughDeriver(); deriver.derive(dependencies).collectAsList(); }