HiveOutput implements BulkOutput, ProvidesAlias, ProvidesValidations { public Dataset<Row> alignColumns(Dataset<Row> input) { Boolean caseSensitive = Contexts.getSparkSession().sparkContext().getConf(). getBoolean(SPARK_SQL_CASE_SENSITIVE_CONFIG, false); Set<String> inputCols = new HashSet<String>(); for (String col : Arrays.asList(input.schema().fieldNames())) { inputCols.add((caseSensitive) ? col : col.toLowerCase()); } List<String> tableCols = new ArrayList<String>(); for (String col : Contexts.getSparkSession().table(tableName).schema().fieldNames()) { tableCols.add((caseSensitive) ? col : col.toLowerCase()); } List<Column> alignedCols = new ArrayList<Column>(); for (String column : tableCols) { alignedCols.add((inputCols.contains(column)) ? functions.col(column) : functions.lit(null).alias(column)); } return input.select(alignedCols.toArray(new Column[alignedCols.size()])); }  @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; }
@Test public void alignColumns() throws Exception { SparkSession spark = Contexts.getSparkSession(); String targetTable = "temp_target_table"; Map<String, Object> paramMap = new HashMap<>(); paramMap.put(HiveOutput.TABLE_CONFIG, targetTable); config = ConfigFactory.parseMap(paramMap); HiveOutput hiveOutput = new HiveOutput(); assertNoValidationFailures(hiveOutput, config); hiveOutput.configure(config); StructType targetSchema = DataTypes.createStructType(Lists.newArrayList( DataTypes.createStructField("zip_code", DataTypes.IntegerType, true), DataTypes.createStructField("city", DataTypes.StringType, true), DataTypes.createStructField("state", DataTypes.StringType, true), DataTypes.createStructField("Fname", DataTypes.StringType, true), DataTypes.createStructField("lname", DataTypes.StringType, true) )); spark.createDataFrame(spark.emptyDataFrame().rdd(), targetSchema) .createOrReplaceTempView(targetTable); String stepDef = "SELECT \"foo\" AS fname, \"bar\" AS lname, 18 AS age, 12345 AS ZIP_CODE"; Dataset<Row> stepDataframe = spark.sql(stepDef); Dataset<Row> alignedDataframe = hiveOutput.alignColumns(stepDataframe); Row alignedData = alignedDataframe.first(); String expectedDef = "SELECT 12345 AS zip_code, NULL as city, NULL as state, " + "\"foo\" AS fname, \"bar\" AS lname"; Row expectedData = spark.sql(expectedDef).first(); assertEquals(expectedData.size(), alignedData.size()); for (int i = 0; i < alignedData.length(); i++) { assertEquals(alignedData.get(i), expectedData.get(i)); } assertEquals(Arrays.asList(alignedData.schema().fieldNames()), Arrays.asList(expectedData.schema().fieldNames())); }