如何使用sqlContext计算累计总和

Question

我知道我们可以在pyspark中使用Window函数来计算累积和。但是Window仅在HiveContext中受支持，而在SQLContext中不受支持。我需要使用SQLContext，因为HiveContext不能在多进程中运行。

有没有什么有效的方法可以使用SQLContext计算累积和？一个简单的方法是将数据加载到驱动程序的内存中并使用numpy.cumsum，但问题是数据需要能够放入内存

Answer 1

不确定这是否是您想要的，但这里有两个如何使用sqlContext计算累积总和的示例：

首先，当你想按某些类别划分时：

from pyspark.sql.types import StructType, StringType, LongType
from pyspark.sql import SQLContext
rdd = sc.parallelize([
    ("Tablet", 6500), 
    ("Tablet", 5500), 
    ("Cell Phone", 6000), 
    ("Cell Phone", 6500), 
    ("Cell Phone", 5500)
    ])
schema = StructType([
    StructField("category", StringType(), False),
    StructField("revenue", LongType(), False)
    ])
df = sqlContext.createDataFrame(rdd, schema)
df.registerTempTable("test_table")
df2 = sqlContext.sql("""
SELECT
    category,
    revenue,
    sum(revenue) OVER (PARTITION BY category ORDER BY revenue) as cumsum
FROM
test_table
""")

输出：

[Row(category='Tablet', revenue=5500, cumsum=5500),
 Row(category='Tablet', revenue=6500, cumsum=12000),
 Row(category='Cell Phone', revenue=5500, cumsum=5500),
 Row(category='Cell Phone', revenue=6000, cumsum=11500),
 Row(category='Cell Phone', revenue=6500, cumsum=18000)]

第二，当你只想取一个变量的总和时。将df2更改为：

df2 = sqlContext.sql("""
SELECT
    category,
    revenue,
    sum(revenue) OVER (ORDER BY revenue, category) as cumsum
FROM
test_table
""")

输出：

[Row(category='Cell Phone', revenue=5500, cumsum=5500),
 Row(category='Tablet', revenue=5500, cumsum=11000),
 Row(category='Cell Phone', revenue=6000, cumsum=17000),
 Row(category='Cell Phone', revenue=6500, cumsum=23500),
 Row(category='Tablet', revenue=6500, cumsum=30000)]

希望这能有所帮助。在收集数据后，使用np.cumsum不是很有效，尤其是在数据集很大的情况下。您可以探索的另一种方法是使用简单的RDD转换，如groupByKey（），然后使用map通过某个键计算每个组的累积和，然后在最后将其减少。

Answer 2

这里有一个简单的例子：

import pyspark
from pyspark.sql import window
import pyspark.sql.functions as sf

sc = pyspark.SparkContext(appName="test")
sqlcontext = pyspark.SQLContext(sc)
data = sqlcontext.createDataFrame([("Bob", "M", "Boston", 1, 20),
                                   ("Cam", "F", "Cambridge", 1, 25),
                                  ("Lin", "F", "Cambridge", 1, 25),
                                  ("Cat", "M", "Boston", 1, 20),
                                  ("Sara", "F", "Cambridge", 1, 15),
                                  ("Jeff", "M", "Cambridge", 1, 25),
                                  ("Bean", "M", "Cambridge", 1, 26),
                                  ("Dave", "M", "Cambridge", 1, 21),], 
                                 ["name", 'gender', "city", 'donation', "age"])

data.show()

给出输出

+----+------+---------+--------+---+
|name|gender|     city|donation|age|
+----+------+---------+--------+---+
| Bob|     M|   Boston|       1| 20|
| Cam|     F|Cambridge|       1| 25|
| Lin|     F|Cambridge|       1| 25|
| Cat|     M|   Boston|       1| 20|
|Sara|     F|Cambridge|       1| 15|
|Jeff|     M|Cambridge|       1| 25|
|Bean|     M|Cambridge|       1| 26|
|Dave|     M|Cambridge|       1| 21|
+----+------+---------+--------+---+

定义窗口

win_spec = (window.Window
                  .partitionBy(['gender', 'city'])
                  .rowsBetween(window.Window.unboundedPreceding, 0))

#窗口。Window.unboundedPreviding--组的第一行#.rowsBetween（…，0）--0指当前行，如果改为指定-2，则在当前行之前最多2行

现在，这里有一个陷阱：

temp = data.withColumn('cumsum',sum(data.donation).over(win_spec))

错误：

TypeErrorTraceback (most recent call last)
<ipython-input-9-b467d24b05cd> in <module>()
----> 1 temp = data.withColumn('cumsum',sum(data.donation).over(win_spec))
/Users/mupadhye/spark-2.1.0-bin-hadoop2.7/python/pyspark/sql/column.pyc in __iter__(self)
    238 
    239     def __iter__(self):
--> 240         raise TypeError("Column is not iterable")
    241 
    242     # string methods
TypeError: Column is not iterable

这是由于使用了python的sum函数而不是pyspark's。解决此问题的方法是使用pyspark.sql.functions.sum:中的sum函数

temp = data.withColumn('AgeSum',sf.sum(data.donation).over(win_spec))
temp.show()

将给出：

+----+------+---------+--------+---+--------------+
|name|gender|     city|donation|age|CumSumDonation|
+----+------+---------+--------+---+--------------+
|Sara|     F|Cambridge|       1| 15|             1|
| Cam|     F|Cambridge|       1| 25|             2|
| Lin|     F|Cambridge|       1| 25|             3|
| Bob|     M|   Boston|       1| 20|             1|
| Cat|     M|   Boston|       1| 20|             2|
|Dave|     M|Cambridge|       1| 21|             1|
|Jeff|     M|Cambridge|       1| 25|             2|
|Bean|     M|Cambridge|       1| 26|             3|
+----+------+---------+--------+---+--------------+

Answer 3

在尝试解决类似问题后，我使用此代码解决了问题。不确定我是否遗漏了OP的一部分，但这是对SQLContext列求和的一种方法：

from pyspark.conf import SparkConf
from pyspark.context import SparkContext
from pyspark.sql.context import SQLContext
sc = SparkContext() 
sc.setLogLevel("ERROR")
conf = SparkConf()
conf.setAppName('Sum SQLContext Column')
conf.set("spark.executor.memory", "2g")
sqlContext = SQLContext(sc)
def sum_column(table, column):
    sc_table = sqlContext.table(table)
    return sc_table.agg({column: "sum"})
sum_column("db.tablename", "column").show()

Answer 4

windows函数只适用于HiveContext是不正确的。您甚至可以将它们与sqlContext:一起使用

from pyspark.sql.window import *
myPartition=Window.partitionBy(['col1','col2','col3'])
temp= temp.withColumn("#dummy",sum(temp.col4).over(myPartition))