@Test public void testFlatten() { FlatteningTransformer transformer = new FlatteningTransformer(); StructField[] nestedStructFields = new StructField[] {new StructField("nestedIntColumn", DataTypes.IntegerType, true, Metadata.empty()), new StructField("nestedStringColumn", DataTypes.StringType, true, Metadata.empty()),}; StructField[] structFields = new StructField[] {new StructField("intColumn", DataTypes.IntegerType, true, Metadata.empty()), new StructField("stringColumn", DataTypes.StringType, true, Metadata.empty()), new StructField("nestedStruct", DataTypes.createStructType(nestedStructFields), true, Metadata.empty())}; StructType schema = new StructType(structFields); String flattenedSql = transformer.flattenSchema(schema, null); assertEquals("intColumn as intColumn,stringColumn as stringColumn," + "nestedStruct.nestedIntColumn as nestedStruct_nestedIntColumn," + "nestedStruct.nestedStringColumn as nestedStruct_nestedStringColumn", flattenedSql); }
public String flattenSchema(StructType schema, String prefix) { final StringBuilder selectSQLQuery = new StringBuilder(); for (StructField field : schema.fields()) { final String fieldName = field.name(); final String colName = prefix == null ? fieldName : (prefix + "." + fieldName); if (field.dataType().getClass().equals(StructType.class)) { selectSQLQuery.append(flattenSchema((StructType) field.dataType(), colName)); } else { selectSQLQuery.append(colName); selectSQLQuery.append(" as "); selectSQLQuery.append(colName.replace(".", "_")); } selectSQLQuery.append(","); } if (selectSQLQuery.length() > 0) { selectSQLQuery.deleteCharAt(selectSQLQuery.length() - 1); } return selectSQLQuery.toString(); }
FlatteningTransformer implements Transformer { public String flattenSchema(StructType schema, String prefix) { final StringBuilder selectSQLQuery = new StringBuilder(); for (StructField field : schema.fields()) { final String fieldName = field.name(); final String colName = prefix == null ? fieldName : (prefix + "." + fieldName); if (field.dataType().getClass().equals(StructType.class)) { selectSQLQuery.append(flattenSchema((StructType) field.dataType(), colName)); } else { selectSQLQuery.append(colName); selectSQLQuery.append(" as "); selectSQLQuery.append(colName.replace(".", "_")); } selectSQLQuery.append(","); } if (selectSQLQuery.length() > 0) { selectSQLQuery.deleteCharAt(selectSQLQuery.length() - 1); } return selectSQLQuery.toString(); } }
FlatteningTransformer implements Transformer { public String flattenSchema(StructType schema, String prefix) { final StringBuilder selectSQLQuery = new StringBuilder(); for (StructField field : schema.fields()) { final String fieldName = field.name(); final String colName = prefix == null ? fieldName : (prefix + "." + fieldName); if (field.dataType().getClass().equals(StructType.class)) { selectSQLQuery.append(flattenSchema((StructType) field.dataType(), colName)); } else { selectSQLQuery.append(colName); selectSQLQuery.append(" as "); selectSQLQuery.append(colName.replace(".", "_")); } selectSQLQuery.append(","); } if (selectSQLQuery.length() > 0) { selectSQLQuery.deleteCharAt(selectSQLQuery.length() - 1); } return selectSQLQuery.toString(); }  }
FlatteningTransformer implements Transformer { public String flattenSchema(StructType schema, String prefix) { final StringBuilder selectSQLQuery = new StringBuilder(); for (StructField field : schema.fields()) { final String fieldName = field.name(); final String colName = prefix == null ? fieldName : (prefix + "." + fieldName); if (field.dataType().getClass().equals(StructType.class)) { selectSQLQuery.append(flattenSchema((StructType) field.dataType(), colName)); } else { selectSQLQuery.append(colName); selectSQLQuery.append(" as "); selectSQLQuery.append(colName.replace(".", "_")); } selectSQLQuery.append(","); } if (selectSQLQuery.length() > 0) { selectSQLQuery.deleteCharAt(selectSQLQuery.length() - 1); } return selectSQLQuery.toString(); }  @Override Dataset<Row> apply(JavaSparkContext jsc, SparkSession sparkSession, Dataset<Row> rowDataset,
      TypedProperties properties); String flattenSchema(StructType schema, String prefix); }
FlatteningTransformer implements Transformer { public String flattenSchema(StructType schema, String prefix) { final StringBuilder selectSQLQuery = new StringBuilder(); for (StructField field : schema.fields()) { final String fieldName = field.name(); final String colName = prefix == null ? fieldName : (prefix + "." + fieldName); if (field.dataType().getClass().equals(StructType.class)) { selectSQLQuery.append(flattenSchema((StructType) field.dataType(), colName)); } else { selectSQLQuery.append(colName); selectSQLQuery.append(" as "); selectSQLQuery.append(colName.replace(".", "_")); } selectSQLQuery.append(","); } if (selectSQLQuery.length() > 0) { selectSQLQuery.deleteCharAt(selectSQLQuery.length() - 1); } return selectSQLQuery.toString(); }  @Override Dataset<Row> apply(JavaSparkContext jsc, SparkSession sparkSession, Dataset<Row> rowDataset,
      TypedProperties properties); String flattenSchema(StructType schema, String prefix);  }