基于此answer 我需要进行一些逐行计算
result= (reduce(add, (<some row wise calculation on col(x)> for x in df.columns[1:])) / n).alias("result")
但在此之前我需要按降序对行值进行排序(更改数据框中每行的列顺序?) 假设我有以下行
3,7,21,9
5,15,10,2
例如,我需要知道每行每个值的排名(顺序),然后计算总和(值/索引)
对于第一行
21 ->4,9->3,7->3,3->1,sum(21/4,9/3,7/3,3/1)
第二行
15->4,10->3,5->2,2->1,sum(15/4,10/4,5/2,2/1)
不是重复的,因为我需要不是按列而是按行排序
最佳答案
假设您的输入数据框如下
+----+----+----+----+
|col1|col2|col3|col4|
+----+----+----+----+
|3 |7 |21 |9 |
|5 |15 |10 |2 |
+----+----+----+----+
然后你可以编写一个udf函数来获取你想要的输出列
from pyspark.sql import functions as f
from pyspark.sql import types as t
def sortAndIndex(list):
return sorted([(value, index+1) for index, value in enumerate(sorted(list))], reverse=True)
sortAndIndexUdf = f.udf(sortAndIndex, t.ArrayType(t.StructType([t.StructField('key', t.IntegerType(), True), t.StructField('value', t.IntegerType(), True)])))
df.withColumn('sortedAndIndexed', sortAndIndexUdf(f.array([x for x in df.columns])))
这应该给你
+----+----+----+----+----------------------------------+
|col1|col2|col3|col4|sortedAndIndexed |
+----+----+----+----+----------------------------------+
|3 |7 |21 |9 |[[21, 4], [9, 3], [7, 2], [3, 1]] |
|5 |15 |10 |2 |[[15, 4], [10, 3], [5, 2], [2, 1]]|
+----+----+----+----+----------------------------------+
更新
您评论为
my calculation should be sum(value/index) so probably using yours udf funcrtion I should return some kind of reduce(add,)?
为此你可以做到
from pyspark.sql import functions as f
from pyspark.sql import types as t
def divideAndSum(list):
return sum([float(value)/(index+1) for index, value in enumerate(sorted(list))])
divideAndSumUdf = f.udf(divideAndSum, t.DoubleType())
df.withColumn('divideAndSum', divideAndSumUdf(f.array([x for x in df.columns])))
这应该给你
+----+----+----+----+------------------+
|col1|col2|col3|col4|divideAndSum |
+----+----+----+----+------------------+
|3 |7 |21 |9 |14.75 |
|5 |15 |10 |2 |11.583333333333334|
+----+----+----+----+------------------+
关于python - Spark 上的行明智计算,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/50666790/