在 PySPARK 中创建一个列，其中包含从所有其他列创建的值作为 JSON

Question

我有如下数据框：

+----------+----------+--------+
|     FNAME|     LNAME|     AGE| 
+----------+----------+--------+
|      EARL|     JONES|      35|
|      MARK|      WOOD|      20|
+----------+----------+--------+

我正在尝试向该数据框添加一个新列作为值，它应该是这样的：

+----------+----------+--------+------+------------------------------------+
|     FNAME|     LNAME|     AGE|                            VALUE          |
+----------+----------+--------+-------------------------------------------+
|      EARL|     JONES|      35|{"FNAME":"EARL","LNAME":"JONES","AGE":"35"}|
|      MARK|      WOOD|      20|{"FNAME":"MARK","WOOD":"JONES","AGE":"20"} |
+----------+----------+--------+-------------------------------------------+

我无法使用 withColumn 或任何 json 函数实现此目的。

我们将不胜感激。

Spark: 2.3
Python: 3.7.x

Answer 1

请考虑使用 SQL 函数 to_json，您可以在 org.apache.spark.sql.functions

中找到它

解决方法如下：

df.withColumn("VALUE", to_json(struct($"FNAME", $"LNAME", $"AGE"))

您还可以避免指定列的名称，如下所示：

df.withColumn("VALUE", to_json(struct(df.columns.map(col): _*)

PS：我提供的代码是用scala写的，但是Python的逻辑是一样的，你只需要使用spark SQL函数就可以了编程语言。

希望对你有帮助，

Answer 2

scala解决方案：

  val df2 = df.select(
    to_json(
      map_from_arrays(lit(df.columns), array('*))
    ).as("value")
  )

pyton 解决方案：（我不知道如何像在 scala 中那样对 n-cols 这样做，因为 map_from_arrays 在 pyspark 中不存在）

    import pyspark.sql.functions as f

    df.select(f.to_json(
        f.create_map(f.lit("FNAME"), df.FNAME, f.lit("LNAME"), df.LNAME, f.lit("AGE"), df.AGE)
        ).alias("value")
    ).show(truncate=False)

输出:

    +-------------------------------------------+
    |value                                      |
    +-------------------------------------------+
    |{"FNAME":"EARL","LNAME":"JONES","AGE":"35"}|
    |{"FNAME":"MARK","LNAME":"WOOD","AGE":"20"} |
    +-------------------------------------------+

Answer 3

使用以下方法实现：

df.withColumn("VALUE", to_json(struct([df[x] for x in df.columns])))

在 PySPARK 中创建一个列，其中包含从所有其他列创建的值作为 JSON

Create a column with values created from all other columns as a JSON in PySPARK

apache-spark

pyspark-dataframes