python - 动态增量值pyspark数据帧-6ren

python - 动态增量值pyspark数据帧

转载作者：太空宇宙更新时间：2023-11-03 14:43:08

我有以下数据框

+--------------------+----------------+----------------+-----------+---+
|         patient_gid|interaction_desc|interaction_date|rx_end_date|rnk|
+--------------------+----------------+----------------+-----------+---+
|00000000000072128380|           prod1|      2009-02-23| 2009-05-22|  1|
|00000000000072128380|           prod1|      2010-04-05| 2009-05-22|  2|
|00000000000072128380|           prod1|      2009-03-23| 2009-05-22|  3|
|00000000000072128380|           prod1|      2009-04-20| 2009-05-22|  4|
|00000000000072128380|           prod1|      2009-05-16| 2009-05-22|  5|
|00000000000072128380|           prod1|      2009-06-17| 2009-05-22|  6|
|00000000000072128380|           prod1|      2009-07-15| 2009-05-22|  7|
|00000000000072128380|           prod1|      2009-08-12| 2009-05-22|  8|
|00000000000072128380|           prod1|      2009-09-05| 2009-05-22|  9|
|00000000000072128380|           prod1|      2009-10-06| 2009-05-22| 10|
|00000000000072128380|           prod2|      2009-10-28| 2009-05-22|  1|
|00000000000072128380|           prod2|      2009-12-02| 2009-05-22|  2|
|00000000000072128380|           prod2|      2010-05-10| 2009-05-22|  3|
|00000000000072128380|           prod2|      2008-05-22| 2009-05-22|  4|
|00000000000072128380|           prod2|      2010-07-06| 2009-05-22|  5|
|00000000000072128380|           prod2|      2010-08-03| 2009-05-22|  6|
|00000000000072128380|           prod2|      2010-09-23| 2009-05-22|  7|
|00000000000072128380|           prod2|      2010-10-20| 2009-05-22|  8|
|00000000000072128380|           prod2|      2010-01-29| 2009-05-22|  9|
|00000000000072128380|           prod2|      2008-05-22| 2009-05-22| 10|
+--------------------+----------------+----------------+-----------+---+

用例:我想添加具有以下逻辑的新专栏剧集如果排名为 1 Episode =1 。如果排名 > 1 并且乘积相同且交互日期 > rx_end_date 则剧集 = 上一集 + 1，否则剧集 = 上一集

预期结果是

+--------------------+----------------+----------------+-----------+---+-------+
|         patient_gid|interaction_desc|interaction_date|rx_end_date|rnk|episode|
+--------------------+----------------+----------------+-----------+---+-------+
|00000000000072128380|           prod1|      2009-02-23| 2009-05-22|  1|      1|
|00000000000072128380|           prod1|      2010-04-05| 2009-05-22|  2|      2|
|00000000000072128380|           prod1|      2009-03-23| 2009-05-22|  3|      2|
|00000000000072128380|           prod1|      2009-04-20| 2009-05-22|  4|      2|
|00000000000072128380|           prod1|      2009-05-16| 2009-05-22|  5|      2|
|00000000000072128380|           prod1|      2009-06-17| 2009-05-22|  6|      3|
|00000000000072128380|           prod1|      2009-07-15| 2009-05-22|  7|      4|
|00000000000072128380|           prod1|      2009-08-12| 2009-05-22|  8|      5|
|00000000000072128380|           prod1|      2009-09-05| 2009-05-22|  9|      6|
|00000000000072128380|           prod1|      2009-10-06| 2009-05-22| 10|      7|
|00000000000072128380|           prod2|      2009-10-28| 2009-05-22|  1|      1|
|00000000000072128380|           prod2|      2009-12-02| 2009-05-22|  2|      2|
|00000000000072128380|           prod2|      2010-05-10| 2009-05-22|  3|      3|
|00000000000072128380|           prod2|      2008-05-22| 2009-05-22|  4|      3|
|00000000000072128380|           prod2|      2010-07-06| 2009-05-22|  5|      4|
|00000000000072128380|           prod2|      2010-08-03| 2009-05-22|  6|      5|
|00000000000072128380|           prod2|      2010-09-23| 2009-05-22|  7|      6|
|00000000000072128380|           prod2|      2010-10-20| 2009-05-22|  8|      7|
|00000000000072128380|           prod2|      2010-01-29| 2009-05-22|  9|      8|
|00000000000072128380|           prod2|      2008-05-22| 2009-05-22| 10|      8|
+--------------------+----------------+----------------+-----------+---+-------+

我想使用 Spark 窗口函数来实现上述逻辑，或者使用任何 Spark 数据框函数来执行此操作？

最佳答案

希望这有帮助!

from pyspark.sql.functions import col, when, lag, last
from pyspark.sql.window import Window
import sys

df = sc.parallelize([
    ['00000000000072128380', 'prod1', '2009-02-23', '2009-05-22', 1],
    ['00000000000072128380', 'prod1', '2010-04-05', '2009-05-22', 2],
    ['00000000000072128380', 'prod1', '2009-03-23', '2009-05-22', 3],
    ['00000000000072128380', 'prod1', '2009-04-20', '2009-05-22', 4],
    ['00000000000072128380', 'prod1', '2009-05-16', '2009-05-22', 5],
    ['00000000000072128380', 'prod1', '2009-06-17', '2009-05-22', 6],
    ['00000000000072128380', 'prod1', '2009-07-15', '2009-05-22', 7],
    ['00000000000072128380', 'prod1', '2009-08-12', '2009-05-22', 8],
    ['00000000000072128380', 'prod1', '2009-09-05', '2009-05-22', 9],
    ['00000000000072128380', 'prod1', '2009-10-06', '2009-05-22', 10],
    ['00000000000072128380', 'prod2', '2009-10-28', '2009-05-22', 1],
    ['00000000000072128380', 'prod2', '2009-12-02', '2009-05-22', 2],
    ['00000000000072128380', 'prod2', '2010-05-10', '2009-05-22', 3],
    ['00000000000072128380', 'prod2', '2008-05-22', '2009-05-22', 4],
    ['00000000000072128380', 'prod2', '2010-07-06', '2009-05-22',  5],
    ['00000000000072128380', 'prod2', '2010-08-03', '2009-05-22',  6],
    ['00000000000072128380', 'prod2', '2010-09-23', '2009-05-22',  7],
    ['00000000000072128380', 'prod2', '2010-10-20', '2009-05-22',  8],
    ['00000000000072128380', 'prod2', '2010-01-29', '2009-05-22',  9],
    ['00000000000072128380', 'prod2', '2008-05-22', '2009-05-22', 10]]).toDF(('patient_gid','interaction_desc', 'interaction_date', 'rx_end_date', 'rnk'))

w = Window.partitionBy(col("interaction_desc")).orderBy(col("rnk"))
df1 = df.withColumn("episode_temp",
                    when(col('rnk')==1, 1).
                    when((col('rnk')>1) &
                         (col('interaction_desc') == lag("interaction_desc").over(w)) &
                         (col('interaction_date') > col('rx_end_date')), col('rnk')).
                    otherwise(None))
df1 = df1.withColumn("episode", last('episode_temp', True).over(w.rowsBetween(-sys.maxsize, 0))).drop('episode_temp')
df1.show()

输出为

+--------------------+----------------+----------------+-----------+---+-------+
|         patient_gid|interaction_desc|interaction_date|rx_end_date|rnk|episode|
+--------------------+----------------+----------------+-----------+---+-------+
|00000000000072128380|           prod1|      2009-02-23| 2009-05-22|  1|      1|
|00000000000072128380|           prod1|      2010-04-05| 2009-05-22|  2|      2|
|00000000000072128380|           prod1|      2009-03-23| 2009-05-22|  3|      2|
|00000000000072128380|           prod1|      2009-04-20| 2009-05-22|  4|      2|
|00000000000072128380|           prod1|      2009-05-16| 2009-05-22|  5|      2|
|00000000000072128380|           prod1|      2009-06-17| 2009-05-22|  6|      6|
|00000000000072128380|           prod1|      2009-07-15| 2009-05-22|  7|      7|
|00000000000072128380|           prod1|      2009-08-12| 2009-05-22|  8|      8|
|00000000000072128380|           prod1|      2009-09-05| 2009-05-22|  9|      9|
|00000000000072128380|           prod1|      2009-10-06| 2009-05-22| 10|     10|
|00000000000072128380|           prod2|      2009-10-28| 2009-05-22|  1|      1|
|00000000000072128380|           prod2|      2009-12-02| 2009-05-22|  2|      2|
|00000000000072128380|           prod2|      2010-05-10| 2009-05-22|  3|      3|
|00000000000072128380|           prod2|      2008-05-22| 2009-05-22|  4|      3|
|00000000000072128380|           prod2|      2010-07-06| 2009-05-22|  5|      5|
|00000000000072128380|           prod2|      2010-08-03| 2009-05-22|  6|      6|
|00000000000072128380|           prod2|      2010-09-23| 2009-05-22|  7|      7|
|00000000000072128380|           prod2|      2010-10-20| 2009-05-22|  8|      8|
|00000000000072128380|           prod2|      2010-01-29| 2009-05-22|  9|      9|
|00000000000072128380|           prod2|      2008-05-22| 2009-05-22| 10|      9|
+--------------------+----------------+----------------+-----------+---+-------+

输出与您想要的并不完全相同，但相似，更重要的是它是单调递增的。

关于python - 动态增量值pyspark数据帧，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/46448818/

文章推荐： c# - 如何覆盖 Autofac 子范围中的 Serilog 记录器？

文章推荐： symfony - HTTPS 上的网络套接字

文章推荐： python - 无法使用 scrapy 使用此代码提取任何数据

文章推荐： spring - PGDatasource 不会连接到远程服务器

Java 增量 i = i++;
这个问题已经有答案了: What is x after "x = x++"? (18 个回答) 已关闭 6 年前。 public static void main(String[] args)
jQuery循环插件交错/增量
我目前正在使用 jquery 循环插件。我有 3 个不同的幻灯片，它们彼此相邻并同时循环播放。我想做的是先关闭第一张幻灯片，然后是第二张幻灯片，然后是第三张幻灯片。无论如何，我可以通过增量或超时来做到
JavaScript 增量
这个问题在这里已经有了答案: 关闭 11 年前。 Possible Duplicate: ++someVariable Vs. someVariable++ in Javascript 我知道您可以
svn --增量 : How to use it and what it does?
我一直在查看 SVN 手册，但无法找到“svn log”和“svn st”的“--incremental”选项的简单用法示例或解释。我正在编写一个开源 SVN GUI 前端，因此我需要一些有关此标志
r - 从矩阵的对角线计算平均值 + 增量
我有这种矩阵。非常抱歉，我没有可重现的示例。表 1: [,1][,2][,3][,4][,5][,6][,7][,8][,9][,10] [1,] 3 NA NA NA
dataframe - 使用数据框在Spark中处理数据差异(增量)
我在hdfs中有一个 Parquet 文件作为我的数据的初始加载。接下来的所有拼花地板只是这些数据集每天都会更改为初始负载(按时间顺序)。这是我的三角洲。我想读取全部或部分 Parquet 文件，以
Javascript Ceil 增量
我目前有这样的功能，可以将任何输入数字四舍五入到最接近的模糊整数值: $(function(){ $('#my_value').blur(function() { $(this).va
c - 如何设计循环数列(增量)？
已关闭。此问题不符合Stack Overflow guidelines 。目前不接受答案。要求提供代码的问题必须表现出对所解决问题的最低限度的了解。包括尝试的解决方案、为什么它们不起作用以及预期结果
java - SQL检查3列+增量
我对 SQL 还很陌生，我想知道我是否可以使用它来自动解决我数据库中的一个复杂问题。也就是说，我每天都在跟踪条目。因此，我们关注的列是: YYYY MM DD XXX YYYY 是年，MM 是月，D
mysql - 数据透视表以外的其他方式 - 增量
我正在开发一个非常简单的数据库，但我不知道数据透视表是否是一个很好的解决方案。如果我使用数据透视表，我需要添加无用的表只是为了增量。让我们从头开始。在用户注册期间，会创建一个新表 GROUP。在G
mysql - 选择查询中的临时变量赋值/增量
在 MySQL 中你可以做这样的事情 SELECT @n := @n + 1 n, first_name, last_name FROM table1, (SELECT
C++增量++运算符重载
如果我正在使用一个类，我知道如何重载运算符 += class temp { public: int i; temp(){ i = 10; } int operator+=(in
Linux Grep 增量
我有两个文件:file1、file2。我想从 file2 中获取 file1 中不存在的行。我读过 post这告诉我使用 grep 的 -v 标志来执行此操作(我阅读了 grep 的手册页，但仍然不
python - 如何使用python日期时间函数/增量？
我看了很多类似的题，功能很简单，用于API的嵌套for循环，每分钟可以调用5次。所以我将一年数据的范围设置为 75。你们能帮我解决这个问题吗？提前致谢! 第一部分正在运行，输入列表中的邮政编码。 fo
python - 计算数据框的每日返回/增量
所以我想计算每日返回/增量的一些时间序列数据，其中每日增量 = value_at_time(T)/value_at_time(T-1) import pandas as pd df=pd.DataFr
c++ - 增量、前增量和后增量
请帮我解决这个问题。该表达式之后的步骤是: //Expression offSpring1[m1++] = temp1; //Steps: 1.- increment m1 2.- assign te
java - 增量/选择性同步远程系统
我正在开发一个解决方案，在该解决方案中，我通过 webapp 不同类型的实体(例如中央数据库上的用户、组、部门信息)和 ldap 进行身份验证。但是最终用户将与来自远程位置(他的办公室、节点)的数据交
python - 获取两个字典列表之间的差异(增量)
我有以下 Python 数据结构: data1 = [{'name': u'String 1'}, {'name': u'String 2'}] data2 = [{'name': u'String
java - AtomicInteger 增量
如果 AtomicInteger 会发生什么？达到 Integer.MAX_VALUE 并递增？值会回到零吗？最佳答案由于integer overflow，它会环绕, 到 Integer.MIN
c - 0.00001 增量
我是 C 的初学者，我正在尝试在 While 循环中进行 0.00001 增量。例如，double t = 0.00001 并且我希望循环每次以 0.00001 的增量运行，第二次是 0.00002

太空宇宙

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城

python - 动态增量值pyspark数据帧