在java spark中更新struct type列中的值

Question

我希望能够更新嵌套数据集中的值。为此，我在Spark中创建了一个嵌套数据集。它有以下模式结构：-

root

 |-- field_a: string (nullable = false)

 |-- field_b: struct (nullable = true)

 |    |-- field_d: struct(nullable = false)
          |-- field_not_to_update: string(nullable = true)

 |        |-- field_to_update: string(nullable = false)
 |   field_c: string (nullable = false)

现在我想更新数据集中field_to_update中的值

aFooData.withColumn("field_b.field_d.field_to_update", lit("updated_val")

也试过，

aFooData.foreach(new ClassWithForEachFunction());

实现ForEach函数的方法

到目前为止，它们都没有取得任何成果，在第二种情况下，我使用foreach获得了相同的数据集，并使用名为field_b.field_d.field_to_update的新列。还有其他类似的建议吗？

Answer 1

更“类似Java”的方法是将数据帧转换为（类型化的）数据集，然后使用映射调用来更改数据。从Java的角度来看，代码很容易处理。但缺点是，对于给定的模式，您需要三个JavaBean类。

Dataset<Bean1> ds = df.as(Encoders.bean(Bean1.class));

Dataset<Bean1> updatedDs = ds.map((MapFunction<Bean1, Bean1>) row -> {
    row.getField_b().getField_d().setField_to_update("updated");
    return row;
}, Encoders.bean(Bean1.class));

三个Bean类

public static class Bean1 implements Serializable {
    private String field_a;
    private Bean2 field_b;
    private String field_c;

    //getters and setters
}

public static class Bean2 implements Serializable {
    private Bean3 field_d;

    //getter and setter
}

public static class Bean3 implements Serializable {
    private String field_not_to_update;
    private String field_to_update;

    //getters and setters
}

Answer 2

你必须重建整个模式，你可以用下面的句子在一个实例中完成。

import org.apache.spark.sql.functions.{lit, struct}

df.select(
  df("field_a"), // keep the fields that don't change
  struct( // the field at first level must be reconstructed
     lit("updated_value") as "field_to_update", // transform or set the new elements
     df("fb.field_not_to_update") as "field_not_to_update" // keep the unchanged sub elements and keep the last name
  ) as "field_b", // and we have to keep the name
  df("field_c")
)

在java中，语法将是相同的

Answer 3

请检查下面的代码。

• 从struct中提取字段
• 更新所需的文件
• 重建结构

scala> df.show(false)
+-------+--------------+
|field_a|field_b       |
+-------+--------------+
|parentA|[srinivas, 20]|
|parentB|[ravi, 30]    |
+-------+--------------+


scala> df.printSchema
root
 |-- field_a: string (nullable = true)
 |-- field_b: struct (nullable = true)
 |    |-- field_to_update: string (nullable = true)
 |    |-- field_not_to_update: integer (nullable = true)


scala> df.select("field_a","field_b.field_to_update","field_b.field_not_to_update").withColumn("field_to_update",lit("updated_val")).select(col("field_a"),struct(col("field_to_update"),col("field_not_to_update")).as("field_b")).show(false)
+-------+-----------------+
|field_a|field_b          |
+-------+-----------------+
|parentA|[updated_val, 20]|
|parentB|[updated_val, 30]|
+-------+-----------------+