java - 为 Spark Rows 定义新架构

Question

我有一个 DataFrame，它的一列包含一串 JSON。到目前为止，我已经实现了Function JavaRDD.map 要求的接口(interface)方法:Function<Row,Row>() 。在此函数中，我正在解析 JSON，并创建一个新行，其附加列来自 JSON 中的值。例如:

原始行:

+------+-----------------------------------+
|  id  |        json                       |
+------+-----------------------------------+
|  1   | {"id":"abcd", "name":"dmux",...}  |
+------------------------------------------+

应用我的功能后:

+------+----------+-----------+
|  id  | json_id  | json_name |
+------+----------+-----------+
|  1   | abcd     | dmux      |
+-----------------+-----------+

当我尝试从返回的 JavaRDD 创建新的 DataFrame 时遇到了麻烦。现在我有了这些新行，我需要创建一个架构。该架构高度依赖于 JSON 的结构，因此我试图找出一种将架构数据与 Row 一起从函数传回的方法。目的。我无法使用broadcast变量，因为 SparkContext 不会传递到函数中。

除了在Function的调用者中循环遍历一行中的每一列之外我有什么选择？

Answer 1

您可以创建一个StructType。这是 Scala，但其工作方式相同:

val newSchema = StructType(Array(
  StructField("id", LongType, false),
  StructField("json_id", StringType, false),
  StructField("json_name", StringType, false)
))

val newDf = sqlContext.createDataFrame(rdd, newSchema)

顺便说一句，您需要确保您的 rdd 类型为 RDD[Row]。