python - PySpark DataFrame 中的行与其前导之间的差异 3 行
问题描述
我有一个 CSV 文件,该文件已通过以下代码作为数据框导入:
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
df = spark.read.csv("name of file.csv", inferSchema = True, header = True)
df.show()
输出
+-----+------+-----+
|col1 | col2 | col3|
+-----+------+-----+
| A | 2 | 4 |
+-----+------+-----+
| A | 4 | 5 |
+-----+------+-----+
| A | 7 | 7 |
+-----+------+-----+
| A | 3 | 8 |
+-----+------+-----+
| A | 7 | 3 |
+-----+------+-----+
| B | 8 | 9 |
+-----+------+-----+
| B | 10 | 10 |
+-----+------+-----+
| B | 8 | 9 |
+-----+------+-----+
| B | 20 | 15 |
+-----+------+-----+
我想col4
为col2[n+3]/col2-1
每个组col1
分别创建另一个。
输出应该是
+-----+------+-----+-----+
|col1 | col2 | col3| col4|
+-----+------+-----+-----+
| A | 2 | 4 | 0.5| #(3/2-1)
+-----+------+-----+-----+
| A | 4 | 5 | 0.75| #(7/4-1)
+-----+------+-----+-----+
| A | 7 | 7 | NA |
+-----+------+-----+-----+
| A | 3 | 8 | NA |
+-----+------+-----+-----+
| A | 7 | 3 | NA |
+-----+------+-----+-----+
| B | 8 | 9 | 1.5 |
+-----+------+-----+-----+
| B | 10 | 10 | NA |
+-----+------+-----+-----+
| B | 8 | 9 | NA |
+-----+------+-----+-----+
| B | 20 | 15 | NA |
+-----+------+-----+-----+
我知道如何在 pandas 中执行此操作,但我不确定如何对 PySpark 中的分组列进行一些计算。
目前,我的 PySpark 版本是 2.4
解决方案
我的 Spark 版本是2.2
. 已经使用了lead()和Window() 。供参考。
from pyspark.sql.window import Window
from pyspark.sql.functions import lead, col
my_window = Window.partitionBy('col1').orderBy('col1')
df = df.withColumn('col2_lead_3', lead(col('col2'),3).over(my_window))\
.withColumn('col4',(col('col2_lead_3')/col('col2'))-1).drop('col2_lead_3')
df.show()
+----+----+----+----+
|col1|col2|col3|col4|
+----+----+----+----+
| B| 8| 9| 1.5|
| B| 10| 10|null|
| B| 8| 9|null|
| B| 20| 15|null|
| A| 2| 4| 0.5|
| A| 4| 5|0.75|
| A| 7| 7|null|
| A| 3| 8|null|
| A| 7| 3|null|
+----+----+----+----+
推荐阅读
- python - Matplotlib 的标题和副标题没有正确对齐
- python - 我正在尝试用 Python 开发一个带有 nltk 库的聊天机器人
- python-3.x - 拆分字符串的内容
- sql - SQL Server:通过通用触发器记录数据库更改
- c - C 带取消和回调的定时器
- javascript - Chrome 扩展程序:后台脚本仅第一次正常运行
- angular - 自定义角度组件在其容器之外呈现
- nginx - 具有非捆绑 Nginx 的 GitLab 返回 502 不响应
- c - copy_to/from_user() 的源地址和目标地址来自哪里?
- java - 如何使用 JSON 正文在 java 中发送 post 请求