public String flattenSchema(StructType schema, String prefix) {
    final StringBuilder selectSQLQuery = new StringBuilder();

    for (StructField field : schema.fields()) {
      final String fieldName = field.name();

      // it is also possible to expand arrays by using Spark "expand" function.
      // As it can increase data size significantly we later pass additional property with a
      // list of arrays to expand.
      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();
  }
@Test
  public void testFlatten() {
    FlatteningTransformer transformer = new FlatteningTransformer();

    // Init
    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);
  }