重命名 Scala Spark 2.0 数据集中的嵌套字段

Question

我正在尝试使用 Spark 2.0 重命名案例类的数据集中的嵌套字段。一个例子如下，我试图将 "element" 重命名为 "address" （保持它在数据结构中的嵌套位置）：

df.printSchema
//Current Output:
root
 |-- companyAddresses: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- addressLine: string (nullable = true)
 |    |    |-- addressCity: string (nullable = true)
 |    |    |-- addressCountry: string (nullable = true)
 |    |    |-- url: string (nullable = true)

//Desired Output:
root
 |-- companyAddresses: array (nullable = true)
 |    |-- address: struct (containsNull = true)
 |    |    |-- addressLine: string (nullable = true)
 |    |    |-- addressCity: string (nullable = true)
 |    |    |-- addressCountry: string (nullable = true)
 |    |    |-- url: string (nullable = true)

仅供参考，以下内容无效：

df.withColumnRenamed("companyAddresses.element","companyAddresses.address") 
df.withColumnRenamed("companyAddresses.element","address")

Answer 1

你在这里要求的是不可能的。 companyAddresses 是一个数组， element 根本不是一列。它只是数组成员模式的指示符。它不能 selected，也不能重命名。

您只能重命名父容器：

df.withColumnRenamed("companyAddresses", "foo")

或通过修改架构来命名各个字段。在简单的情况下，也可以使用 struct 和 select:

df.select(struct($"foo".as("bar"), $"bar".as("foo")))

但是这里显然不适用

Answer 2

您可以为此编写一个小的递归函数，并使用映射：

final JavaRDD rdd = df.toJavaRDD().map(row -> ....);


private static void flatDocument(Row input, Map<String,Object> outValues, String fqn)
{
    final StructType schema = input.schema();

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

        String key = fqn == null ? fieldName : fqn + "_" + fieldName;

        Object buffer = input.getAs(fieldName);

        if (field.dataType().getClass().equals(StructType.class))
        {
            if (buffer != null) {
                flatDocument((Row) buffer, outValues, key);
            }
        }
        else
        {
            outValues.put(key, buffer);
        }
    }
}

但是您需要一个模式将其转换回数据集：/

重命名 Scala Spark 2.0 数据集中的嵌套字段

Re-name nested field in Scala Spark 2.0 Dataset

scala

apache-spark

apache-spark-dataset