python - PySpark - 从值列表中添加列

标签 python list apache-spark pyspark apache-spark-sql

我必须根据值列表将列添加到 PySpark 数据帧。

a= spark.createDataFrame([("Dog", "Cat"), ("Cat", "Dog"), ("Mouse", "Cat")],["Animal", "Enemy"])

我有一个名为评级的列表,它是每个宠物的评级。

rating = [5,4,1]

我需要在数据框后面附加一个名为“Rating”的列,这样

+------+-----+------+
|Animal|Enemy|Rating|
+------+-----+------+
|   Dog|  Cat|     5|
|   Cat|  Dog|     4|
| Mouse|  Cat|     1|
+------+-----+------+

我已完成以下操作,但它仅返回评级列列表中的第一个值

def add_labels():
    return rating.pop(0)

labels_udf = udf(add_labels, IntegerType())

new_df = a.withColumn('Rating', labels_udf()).cache()

输出:

+------+-----+------+
|Animal|Enemy|Rating|
+------+-----+------+
|   Dog|  Cat|     5|
|   Cat|  Dog|     5|
| Mouse|  Cat|     5|
+------+-----+------+

最佳答案

from pyspark.sql.functions import monotonically_increasing_id, row_number
from pyspark.sql import Window

#sample data
a= sqlContext.createDataFrame([("Dog", "Cat"), ("Cat", "Dog"), ("Mouse", "Cat")],
                               ["Animal", "Enemy"])
a.show()

#convert list to a dataframe
rating = [5,4,1]
b = sqlContext.createDataFrame([(l,) for l in rating], ['Rating'])

#add 'sequential' index and join both dataframe to get the final result
a = a.withColumn("row_idx", row_number().over(Window.orderBy(monotonically_increasing_id())))
b = b.withColumn("row_idx", row_number().over(Window.orderBy(monotonically_increasing_id())))

final_df = a.join(b, a.row_idx == b.row_idx).\
             drop("row_idx")
final_df.show()

输入:

+------+-----+
|Animal|Enemy|
+------+-----+
|   Dog|  Cat|
|   Cat|  Dog|
| Mouse|  Cat|
+------+-----+

输出是:

+------+-----+------+
|Animal|Enemy|Rating|
+------+-----+------+
|   Cat|  Dog|     4|
|   Dog|  Cat|     5|
| Mouse|  Cat|     1|
+------+-----+------+

关于python - PySpark - 从值列表中添加列,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/48164206/

相关文章:

python - 如何从 Jupyter 在 HDInsight Spark 集群上提交 python wordcount

python - 无法重命名由 .sum() 函数创建的数据框的列名称

list - LISP:力评估

python - 为什么会出现IndexError : list index out of range

java - 如何确定Spark中shuffle分区的最佳数量

python - 为什么 Laravel Redis::scan ('*' ) 返回预期的 key ,但 Redis::keys ('*' ) 没有返回?

ios - SwiftUI 列表颜色背景

python - 多处理 RDD 列表

scala - 线程 "main"java.lang.NoClassDefFoundError : com/fasterxml/jackson/databind/Module in Intellij while running Kafka App? 中出现异常

python - 使用 numpy.random.normal 时如何指定上限和下限