java - Spark-Java : How to replace a column name in Dataset to new name?

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 中的保留字。因此，来自 MongoDB 的 JSON 数据包含一个键 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(...))。