Flatten Spark data frame fields structure, via SQL in Java

flatten.java

class Toto
{
    public void Main()
    {
        final DataFrame source = GetDataFrame();
        final String querySelectSQL = flattenSchema(source.schema(), null);

        source.registerTempTable("source"); 
      
        final DataFrame flattenData = sqlContext.sql("SELECT " + querySelectSQL + " FROM source")
    }
    
    /**
    * Generate SQL to select columns as flat.
    */
    public String flattenSchema(StructType schema, String prefix)
    {
        final StringBuilder selectSQLQuery = new StringBuilder();

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

            if (fieldName.startsWith("@"))
            {
                continue;
            }

            String colName = prefix == null ? fieldName : (prefix + "." + fieldName);
            String colNameTarget = colName.replace(".", "_");

            if (field.dataType().getClass().equals(StructType.class))
            {
                selectSQLQuery.append(flattenSchema((StructType) field.dataType(), colName));
            }
            else
            {
                selectSQLQuery.append(colName);
                selectSQLQuery.append(" as ");
                selectSQLQuery.append(colNameTarget);
            }

            selectSQLQuery.append(",");
        }

        if (selectSQLQuery.length() > 0)
        {
            selectSQLQuery.deleteCharAt(selectSQLQuery.length() - 1);
        }

        return selectSQLQuery.toString();
    }
}

This is a bit more succinct Java code to flatten a schema and return the fields as a List:

  private List<String> flattenSchema(StructType schema) {
    List<String> allFields = new ArrayList<>();
    flattenSchemaHelper(schema, null, allFields);
    return allFields;
  }

  private void flattenSchemaHelper(StructType schema, String prefix, List<String> fields) {
    for (StructField field : schema.fields()) {
      String colName = prefix == null ? field.name() : (prefix + "." + field.name());
      if (field.dataType().getClass().equals(StructType.class))  {
        flattenSchemaHelper((StructType) field.dataType(), colName, fields);
      } else {
        fields.add(colName);
      }
    }
  }