代码之家 › 专栏 › 技术社区 › Masterbuilder

使用窗口函数将前一行与当前行相加

spark-dataframe pyspark apache-spark

Masterbuilder · 技术社区 · 6 年前

我有一个spark数据框,其中,我想根据当前行金额值和基于groupid和id的金额值的前一行和计算一个运行总计。让我输出df

import findspark
findspark.init()
import pyspark 
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
import pandas as pd


 sc = spark.sparkContext
data1 = {'date': {0: '2018-04-03', 1: '2018-04-04', 2: '2018-04-05', 3: '2018-04-06', 4: '2018-04-07'},
         'id': {0: 'id1', 1: 'id2', 2: 'id1', 3: 'id3', 4: 'id2'},
         'group': {0: '1', 1: '1', 2: '1', 3: '2', 4: '1'},
         'amount': {0: 50, 1: 40, 2: 50, 3: 55, 4: 20}}
df1_pd = pd.DataFrame(data1, columns=data1.keys())

df1 = spark.createDataFrame(df1_pd)
df1.show()


+----------+---+-----+------+
|      date| id|group|amount|
+----------+---+-----+------+
|2018-04-03|id1|    1|    50|
|2018-04-04|id2|    1|    40|
|2018-04-05|id1|    1|    50|
|2018-04-06|id3|    2|    55|
|2018-04-07|id2|    1|    20|
+----------+---+-----+------+

我要找的出口

+----------+---+-----+------+---+
|      date| id|group|amount|sum|
+----------+---+-----+------+---+
|2018-04-03|id1|    1|    50|50 |
|2018-04-04|id2|    1|    40|90 |
|2018-04-05|id1|    1|    50|140|
|2018-04-06|id3|    2|    55|55 |
|2018-04-07|id2|    1|    20|160|
+----------+---+-----+------+---+

1 回复 | 直到 6 年前

user9732570 user9732570 6 年前

窗口定义:

from pyspark.sql.window import Window
from pyspark.sql.functions import sum

w = Window.partitionBy("group").orderBy("date").rowsBetween(
    Window.unboundedPreceding,  # Take all rows from the beginning of frame
    Window.currentRow           # To current row
)

总额:

(df1.withColumn("sum", sum("amount").over(w))
    .orderBy("date")   # Sort for easy inspection. Not necessary
    .show())

结果:

+----------+---+-----+------+---+      
|      date| id|group|amount|sum|
+----------+---+-----+------+---+
|2018-04-03|id1|    1|    50| 50|
|2018-04-04|id2|    1|    40| 90|
|2018-04-05|id1|    1|    50|140|
|2018-04-06|id3|    2|    55| 55|
|2018-04-07|id2|    1|    20|160|
+----------+---+-----+------+---+

推荐文章

Leonard · Pyspark:JSON到Pyspark数据帧

1 年前

Fran Arenas · Pyspark collect()方法在Pycharm或控制台中执行测试时给出了不同的顺序

2 年前

titutubs · 有没有一种更有效的方法来为Databricks SQL中的bin值编写代码?

2 年前

markwatson · AWS Glue:如何在输出中添加具有源文件名的列?

6 年前

juamd · 顺序(k,<元组>)RDD

6 年前

Gaurav Gupta · 如何加载多行记录的CSV文件?

6 年前

Jared · 如何在本地模式下运行的pyspark中读取S3?

6 年前

ka_boom · 在pyspark中链接多个groupBy

6 年前

ds_user · 在apache spark中复制记录计数

6 年前

ds_user · 结合类型和子类型的Apache Spark组

6 年前