Spark (Java API) - 向 JSON 列添加新字段

Question

我正在使用 Spark 从 Cassandra 读取行，其中一列是 JSON 对象。这是数据集中的架构和示例行：

root
 |-- attributes: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- name: string (nullable = true)
 |    |    |-- count: integer (nullable = true)
 |    |    |-- score: double (nullable = true)
 |    |    |-- id: integer (nullable = true)
 |-- id: string (nullable = true)
 |-- col1: string (nullable = true)

+--------------------------------------------------------------------------------+--------+---------+
|attributes                                                                      |id      |col1     |
+--------------------------------------------------------------------------------+--------+---------+
|[[usage,2,5.0,12], [price,1,10.0,48], [hair,1,10.0,23737], [curls,1,10.0,30807]]|19400335|val_str_1|
+--------------------------------------------------------------------------------+--------+---------+

使用以下模式从表中加载：

ArrayType schema =
                DataTypes.createArrayType(
                        new StructType(
                                new StructField[] {
                                        new StructField("name", DataTypes.StringType, true, Metadata.empty()),
                                        new StructField("count", DataTypes.IntegerType, true, Metadata.empty()),
                                        new StructField("score", DataTypes.DoubleType, true, Metadata.empty()),
                                        new StructField("id", DataTypes.IntegerType, true, Metadata.empty())}));

是否可以向 attributes 列注入新字段 - 将来自 id 列的值放在根下，并将来自 col1 列的值放在结构内？

Answer 1

您可以尝试使用transform 函数来更新数组中的每个结构元素：

df.withColumn("attributes", expr("transform(attributes, x -> struct(x.name as name, x.count as count, x.score as score, x.id as id, id as root_id, col1 as root_col1))"))

对于spark ，可以先炸开数组列，然后更新groupby再收集列表：

df.withColumn("attr_structs", explode(col("attributes")))
    .withColumn("attr_structs", struct(col("attr_structs.*"), col("id").alias("root_id"), col("col1").alias("root_col1")))
    .groupBy("id", "col1")
    .agg(collect_list(col("attr_structs")).alias("attributes"))