火花行到JSON

Question

我想从Spark v.1.6（使用scala）数据帧创建一个JSON。我知道有一个简单的解决方案，就是做df.toJSON。

但是，我的问题看起来有点不同。例如，考虑具有以下列的数据帧：

|  A  |     B     |  C1  |  C2  |    C3   |
-------------------------------------------
|  1  | test      |  ab  |  22  |  TRUE   |
|  2  | mytest    |  gh  |  17  |  FALSE  |

我想在最后有一个数据帧

|  A  |     B     |                        C                   |
----------------------------------------------------------------
|  1  | test      | { "c1" : "ab", "c2" : 22, "c3" : TRUE }    |
|  2  | mytest    | { "c1" : "gh", "c2" : 17, "c3" : FALSE }   |

其中C是包含C1、C2、C3的JSON。不幸的是，我在编译时不知道数据框是什么样子的（除了始终“固定”的列A和B）。

至于我需要这个的原因：我使用Protobuf发送结果。不幸的是，我的数据帧有时有比预期更多的列，我仍然会通过Protobuf发送这些列，但我不想在定义中指定所有列。

我怎样才能做到这一点？

Answer 1

这里没有JSON解析器，它适应您的架构：

import org.apache.spark.sql.functions.{col, concat, concat_ws, lit}

df.select(
  col(df.columns(0)),
  col(df.columns(1)),
  concat(
    lit("{"), 
    concat_ws(",",df.dtypes.slice(2, df.dtypes.length).map(dt => {
      val c = dt._1;
      val t = dt._2;
      concat(
        lit("\"" + c + "\":" + (if (t == "StringType") "\""; else "")  ),
        col(c),
        lit(if(t=="StringType") "\""; else "") 
      )
    }):_*), 
    lit("}")
  ) as "C"
).collect()

Answer 2

我使用此命令来解决to_json问题：

output_df = (df.select(to_json(struct(col("*"))).alias("content")))

Answer 3

Spark 2.1应该有对这个用例的本地支持(参见#15354)。

import org.apache.spark.sql.functions.to_json
df.select(to_json(struct($"c1", $"c2", $"c3")))