pyspark - 如何在 Pyspark 中将列表拆分为多列?

标签 pyspark apache-spark-sql

我有:

key   value
a    [1,2,3]
b    [2,3,4]

我想要:
key value1 value2 value3
a     1      2      3
b     2      3      4

似乎在scala中我可以写:df.select($"value._1", $"value._2", $"value._3") ,但在 python 中是不可能的。

那么有没有好的方法可以做到这一点?

最佳答案

这取决于您的“列表”的类型:

  • 如果它是类型 ArrayType() :

    df = hc.createDataFrame(sc.parallelize([['a', [1,2,3]], ['b', [2,3,4]]]), ["key", "value"])
    df.printSchema()
    df.show()
    root
     |-- key: string (nullable = true)
     |-- value: array (nullable = true)
     |    |-- element: long (containsNull = true)
    

    您可以使用 [] 像使用 python 一样访问这些值:

    df.select("key", df.value[0], df.value[1], df.value[2]).show()
    +---+--------+--------+--------+
    |key|value[0]|value[1]|value[2]|
    +---+--------+--------+--------+
    |  a|       1|       2|       3|
    |  b|       2|       3|       4|
    +---+--------+--------+--------+
    
    +---+-------+
    |key|  value|
    +---+-------+
    |  a|[1,2,3]|
    |  b|[2,3,4]|
    +---+-------+
    
  • 如果它是类型 StructType() :(也许你是通过读取 JSON 来构建数据框的)

    df2 = df.select("key", psf.struct(
            df.value[0].alias("value1"), 
            df.value[1].alias("value2"), 
            df.value[2].alias("value3")
        ).alias("value"))
    df2.printSchema()
    df2.show()
    root
     |-- key: string (nullable = true)
     |-- value: struct (nullable = false)
     |    |-- value1: long (nullable = true)
     |    |-- value2: long (nullable = true)
     |    |-- value3: long (nullable = true)
    
    +---+-------+
    |key|  value|
    +---+-------+
    |  a|[1,2,3]|
    |  b|[2,3,4]|
    +---+-------+
    

    您可以使用 * 直接“拆分”列:

    df2.select('key', 'value.*').show()
    +---+------+------+------+
    |key|value1|value2|value3|
    +---+------+------+------+
    |  a|     1|     2|     3|
    |  b|     2|     3|     4|
    +---+------+------+------+
    
  • 关于pyspark - 如何在 Pyspark 中将列表拆分为多列?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45789489/

    相关文章:

    python - 如何在条件下在pyspark上创建新列?

    apache-spark - 如何在 Spark 结构化流中使用流数据帧更新静态数据帧

    scala - 在 Spark Scala 中映射和删除重复项?

    apache-spark - 计算Spark Dataframe中分组数据的分位数

    python - pyspark从RDD中过滤列表

    python - 在同一 IDE 中使用 Spark 和 Python

    java - Spark - 流数据帧/数据集不支持非基于时间的窗口;

    apache-spark - Spark中使用Hive数据库

    python - pyspark 将数据帧写入 hdfs 失败

    python - 使用 python 集的 AggregateBykey