Спарк сглаживает набор данных - PullRequest
Новая
       8

Спарк сглаживает набор данных

0 голосов
/ 20 марта 2020

У меня есть СДР со схемой - 

Schema: {
  "type" : "struct",
  "fields" : [ 

 {
    "name" : "cola",
    "type" : "string",
    "nullable" : true,
    "metadata" : { }
  }, {
    "name" : "mappedcol",
    "type" : {
      "type" : "map",
      "keyType" : "string",
      "valueType" : "string",
      "valueContainsNull" : true
    },
    "nullable" : true,
    "metadata" : { }
  }, {
    "name" : "colc",
    "type" : "string",
    "nullable" : true,
    "metadata" : { }
  }]
  }

Пример значения: 

{
cola : A1,
mappedcol : { mapped1: M1, mapped2: M2, mapped3: M3  }
colc : C1
}

Я хочу вытащить ключи в сопоставленных протоколах на один уровень вверх. В основном сглаживают все столбцы на одном уровне. 

cola, mapped1, mapped2, mapped3, colc
A1, M1,M2,M3, C1

Есть ли элегантный способ сделать это в Java?

1 Ответ

0 голосов
/ 22 марта 2020

Можно получить доступ к отдельным элементам вложенной структуры с точечным синтаксисом, например, select mappedcol.mapped1 вернет M1. Идея состоит в том, чтобы преобразовать схему в список имен столбцов, используя этот синтаксис точек: 

private static List<String> structToColNames(StructField[] fields, String prefix) {
    List<String> columns = new ArrayList<>();
    for( StructField field: fields) {
        String fieldname = field.name();
        if( field.dataType() instanceof StructType) {
            columns.addAll(
                structToColNames(((StructType)field.dataType()).fields(), 
                    prefix + fieldname + "."));
        }
        else {
            columns.add(prefix + fieldname);
        }
    }
    return columns;
}

Результат этой функции затем можно использовать для выбора данных: 

Dataset<Row> df = spark.read().json(<path to json>);
StructField[] fields = df.schema().fields();
List<String> colNames = structToColNames(fields, "");
System.out.println(colNames);
Column[] columns = colNames.stream().map(s -> col(s)).toArray(Column[]::new);
df.select(columns).show();

отпечатки 

[cola, colc, mappedcol.mapped1, mappedcol.mapped2, mappedcol.mapped3]

+----+----+-------+-------+-------+
|cola|colc|mapped1|mapped2|mapped3|
+----+----+-------+-------+-------+
|  A1|  C1|     M1|     M2|     M3|
+----+----+-------+-------+-------+
...