Spark-Java: Как заменить имя столбца в наборе данных <Row>на новое имя?

Question

Я читаю данные из MongoDB, используя Spark в качестве JSON:

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
....
..    
Dataset<Row> ds = MongoSpark.load(jsc).toDF(); 

Мне нужно сохранить этот DF в Hive.
Проблема в том, что одно из имен столбцов вMongoDB - это Timestamp, которое является зарезервированным словом в Hive.Итак, JSON-данные из MongoDB содержат один ключ как timestamp.
Мне нужно заменить этот ключ json "timestamp" на "timestamp_".
Как заменить имя столбца "timestamp" в Dataset<Row> ds до "timestamp_"?

Answer 1

Если вы хотите переименовать вложенный столбец, вы можете сделать следующее:

1. сгладить / разнести все столбцы структуры
2. переименовать столбец
3. свернутьвсе столбцы назад

Допустим, ваша схема набора данных выглядит следующим образом:

root  
      |-- col1
      |-- col2
      |-- struct1
      |    |-- timestamp
      |    |-- a  
      |    |-- b

Так что вы можете сделать следующее

ds = ds
        .select(col("*"), col("struct1.*"))
        .withColumnRenamed("timestamp", "timestamp_")
        .select(
                 col("col1"),
                 col("col2"),
                 struct("timestamp_", "a", "b").as("struct1")
        );

В случае, если вы хотитечтобы переименовать вложенный столбец внутри массива, сначала необходимо расширить массив с помощью функции explode.

ds = ds.select(col("a"), col("b"), explode(col("struct1")))

Затем вы можете переименовать вложенный столбец, как описано выше.Если после переименования вы хотите свернуть структуры обратно под массив, используйте groupBy(...).agg(collect_list(...)).