我正在使用 PySpark，我有一个带有一堆数字列的 Spark 数据框。我想添加一个列，它是所有其他列的总和。
假设我的数据框有“a”、“b”和“c”列。我知道我可以这样做：
df.withColumn('total_col', df.a + df.b + df.c)
问题是我不想单独输入每一列并添加它们，特别是如果我有很多列。我希望能够自动执行此操作，或者通过指定要添加的列名列表来执行此操作。还有另一种方法可以做到这一点吗？
问问题
91148 次
8 回答
50
这并不明显。我看不到 spark Dataframes API 中定义的列的基于行的总和。
版本 2
这可以通过一种相当简单的方式完成：
newdf = df.withColumn('total', sum(df[col] for col in df.columns))
df.columns由 pyspark 作为字符串列表提供，给出了 Spark Dataframe 中的所有列名。对于不同的总和，您可以提供任何其他列名列表。
我没有尝试将此作为我的第一个解决方案，因为我不确定它会如何表现。但它有效。
版本 1
这过于复杂，但也有效。
你可以这样做：
- 用于
df.columns获取列名的列表 - 使用该名称列表来制作列列表
- 将该列表传递给将以折叠类型函数方式调用列的重载添加函数的东西
使用 python 的reduce，了解运算符重载的工作原理，以及此处列的 pyspark 代码变为：
def column_add(a,b):
return a.__add__(b)
newdf = df.withColumn('total_col',
reduce(column_add, ( df[col] for col in df.columns ) ))
注意这是一个python reduce，而不是spark RDD reduce，第二个参数中的括号项需要括号，因为它是一个列表生成器表达式。
经测试，有效！
$ pyspark
>>> df = sc.parallelize([{'a': 1, 'b':2, 'c':3}, {'a':8, 'b':5, 'c':6}, {'a':3, 'b':1, 'c':0}]).toDF().cache()
>>> df
DataFrame[a: bigint, b: bigint, c: bigint]
>>> df.columns
['a', 'b', 'c']
>>> def column_add(a,b):
... return a.__add__(b)
...
>>> df.withColumn('total', reduce(column_add, ( df[col] for col in df.columns ) )).collect()
[Row(a=1, b=2, c=3, total=6), Row(a=8, b=5, c=6, total=19), Row(a=3, b=1, c=0, total=4)]
于 2015-08-12T03:55:56.193 回答
17
最直接的方法是使用expr函数
from pyspark.sql.functions import *
data = data.withColumn('total', expr("col1 + col2 + col3 + col4"))
于 2019-02-08T19:05:58.300 回答
11
解决方案
newdf = df.withColumn('total', sum(df[col] for col in df.columns))
@Paul 发布的作品。尽管如此，我还是遇到了错误，正如我所看到的那样，
TypeError: 'Column' object is not callable
一段时间后，我发现了问题（至少在我的情况下）。问题是我之前用该行导入了一些 pyspark 函数
from pyspark.sql.functions import udf, col, count, sum, when, avg, mean, min
因此该行导入了sumpyspark 命令，而df.withColumn('total', sum(df[col] for col in df.columns))应该使用普通的 pythonsum函数。
您可以使用 删除 pyspark 函数的引用del sum。
否则在我的情况下，我将导入更改为
import pyspark.sql.functions as F
然后将函数引用为F.sum.
于 2018-12-20T11:49:10.210 回答
9
将列表中的多列汇总为一列
PySpark 的sum函数不支持列添加。这可以使用expr函数来实现。
from pyspark.sql.functions import expr
cols_list = ['a', 'b', 'c']
# Creating an addition expression using `join`
expression = '+'.join(cols_list)
df = df.withColumn('sum_cols', expr(expression))
这为我们提供了所需的列总和。
于 2020-04-23T05:55:39.927 回答
1
我的问题与上述类似（更复杂一点），因为我必须在 PySpark 数据框中添加连续的列总和作为新列。此方法使用上述 Paul 版本 1 中的代码：
import pyspark
from pyspark.sql import SparkSession
import pandas as pd
spark = SparkSession.builder.appName('addColAsCumulativeSUM').getOrCreate()
df=spark.createDataFrame(data=[(1,2,3),(4,5,6),(3,2,1)\
,(6,1,-4),(0,2,-2),(6,4,1)\
,(4,5,2),(5,-3,-5),(6,4,-1)]\
,schema=['x1','x2','x3'])
df.show()
+---+---+---+
| x1| x2| x3|
+---+---+---+
| 1| 2| 3|
| 4| 5| 6|
| 3| 2| 1|
| 6| 1| -4|
| 0| 2| -2|
| 6| 4| 1|
| 4| 5| 2|
| 5| -3| -5|
| 6| 4| -1|
+---+---+---+
colnames=df.columns
添加累积和（连续）的新列：
for i in range(0,len(colnames)):
colnameLst= colnames[0:i+1]
colname = 'cm'+ str(i+1)
df = df.withColumn(colname, sum(df[col] for col in colnameLst))
df.show()
+---+---+---+---+---+---+
| x1| x2| x3|cm1|cm2|cm3|
+---+---+---+---+---+---+
| 1| 2| 3| 1| 3| 6|
| 4| 5| 6| 4| 9| 15|
| 3| 2| 1| 3| 5| 6|
| 6| 1| -4| 6| 7| 3|
| 0| 2| -2| 0| 2| 0|
| 6| 4| 1| 6| 10| 11|
| 4| 5| 2| 4| 9| 11|
| 5| -3| -5| 5| 2| -3|
| 6| 4| -1| 6| 10| 9|
+---+---+---+---+---+---+
添加的“累计和”列如下：
cm1 = x1
cm2 = x1 + x2
cm3 = x1 + x2 + x3
于 2018-07-04T14:56:20.713 回答
1
df = spark.createDataFrame([("linha1", "valor1", 2), ("linha2", "valor2", 5)], ("Columna1", "Columna2", "Columna3"))
df.show()
+--------+--------+--------+
|Columna1|Columna2|Columna3|
+--------+--------+--------+
| linha1| valor1| 2|
| linha2| valor2| 5|
+--------+--------+--------+
df = df.withColumn('DivisaoPorDois', df[2]/2)
df.show()
+--------+--------+--------+--------------+
|Columna1|Columna2|Columna3|DivisaoPorDois|
+--------+--------+--------+--------------+
| linha1| valor1| 2| 1.0|
| linha2| valor2| 5| 2.5|
+--------+--------+--------+--------------+
df = df.withColumn('Soma_Colunas', df[2]+df[3])
df.show()
+--------+--------+--------+--------------+------------+
|Columna1|Columna2|Columna3|DivisaoPorDois|Soma_Colunas|
+--------+--------+--------+--------------+------------+
| linha1| valor1| 2| 1.0| 3.0|
| linha2| valor2| 5| 2.5| 7.5|
+--------+--------+--------+--------------+------------+
于 2019-10-30T18:20:16.770 回答
0
一个非常简单的方法是只使用 select 而不是 withcolumn，如下所示：
df = df.select('*', (col("a")+col("b")+col('c).alias("total"))
这应该为您提供所需的总和，并根据要求进行细微更改
于 2020-01-22T14:33:05.207 回答
0
以下方法对我有用：
- 从 pyspark.sql导入 pyspark sql 函数
导入函数为 F - 使用 F.expr(list_of_columns)
data_frame.withColumn('Total_Sum',F.expr('col_name 1 +col_name 2 +..col_name n )
于 2020-08-14T09:08:52.730 回答