PySpark Dataframe 将两列转换为基于第三列值的新元组列

2023-12-05

正如主题所描述的，我有一个 PySpark Dataframe，我需要将两列转换为一个新列，它是基于第三列值的元组列表。该演员阵容将减少或通过键值（本例中为产品 id）展平数据框，结果为一行每个键。

该数据框中有数亿行，有 3700 万个唯一的产品 ID。因此我需要一种在 Spark 集群上进行转换而不带回任何数据的方法到驱动程序（在本例中为 Jupyter）。

以下是我的 1 个产品的数据框摘录：

+-----------+-------------------+-------------+--------+----------+---------------+
| product_id|      purchase_date|days_warranty|store_id|year_month|       category|
+-----------+-------------------+-----------+----------+----------+---------------+
|02147465400|2017-05-16 00:00:00|           30|     205|   2017-05|     CATEGORY A|
|02147465400|2017-04-15 00:00:00|           30|     205|   2017-04|     CATEGORY A|
|02147465400|2018-07-11 00:00:00|           30|     205|   2018-07|     CATEGORY A|
|02147465400|2017-06-14 00:00:00|           30|     205|   2017-06|     CATEGORY A|
|02147465400|2017-03-16 00:00:00|           30|     205|   2017-03|     CATEGORY A|
|02147465400|2017-08-14 00:00:00|           30|     205|   2017-08|     CATEGORY A|
|02147465400|2017-09-12 00:00:00|           30|     205|   2017-09|     CATEGORY A|
|02147465400|2017-01-21 00:00:00|           30|     205|   2017-01|     CATEGORY A|
|02147465400|2018-08-14 00:00:00|           30|     205|   2018-08|     CATEGORY A|
|02147465400|2018-08-23 00:00:00|           30|     205|   2018-08|     CATEGORY A|
|02147465400|2017-10-11 00:00:00|           30|     205|   2017-10|     CATEGORY A|
|02147465400|2017-12-12 00:00:00|           30|     205|   2017-12|     CATEGORY A|
|02147465400|2017-02-15 00:00:00|           30|     205|   2017-02|     CATEGORY A|
|02147465400|2018-04-12 00:00:00|           30|     205|   2018-04|     CATEGORY A|
|02147465400|2018-03-12 00:00:00|           30|     205|   2018-03|     CATEGORY A|
|02147465400|2018-05-15 00:00:00|           30|     205|   2018-05|     CATEGORY A|
|02147465400|2018-02-12 00:00:00|           30|     205|   2018-02|     CATEGORY A|
|02147465400|2018-06-14 00:00:00|           30|     205|   2018-06|     CATEGORY A|
|02147465400|2018-01-11 00:00:00|           30|     205|   2018-01|     CATEGORY A|
|02147465400|2017-07-20 00:00:00|           30|     205|   2017-07|     CATEGORY A|
|02147465400|2017-11-11 00:00:00|           30|     205|   2017-11|     CATEGORY A|
|02147465400|2017-01-05 00:00:00|           90|     205|   2017-01|     CATEGORY B|
|02147465400|2017-01-21 00:00:00|           90|     205|   2017-01|     CATEGORY B|
|02147465400|2017-10-09 00:00:00|           90|     205|   2017-10|     CATEGORY B|
|02147465400|2018-07-11 00:00:00|           90|     205|   2018-07|     CATEGORY B|
|02147465400|2017-04-16 00:00:00|           90|     205|   2017-04|     CATEGORY B|
|02147465400|2018-09-16 00:00:00|           90|     205|   2018-09|     CATEGORY B|
|02147465400|2018-04-14 00:00:00|           90|     205|   2018-04|     CATEGORY B|
|02147465400|2018-01-12 00:00:00|           90|     205|   2018-01|     CATEGORY B|
|02147465400|2017-07-15 00:00:00|           90|     205|   2017-07|     CATEGORY B|
+-----------+-------------------+-----------+----------+----------+---------------+

这是所需的结果数据框，一个产品的一行，其中行原始数据帧的 buy_date 和 days_warranty 列已转换作为基于类别列值的元组数组到新列中：

+-----------+----------------------------+----------------------------+
| product_id|                  CATEGORY A|                  CATEGORY B| 
+-----------+----------------------------+----------------------------+
|02147465400| [ (2017-05-16 00:00:00,30),| [ (2017-01-05 00:00:00,90),| 
|           |   (2017-04-15 00:00:00,30),|   (2017-01-21 00:00:00,90),|
|           |   (2018-07-11 00:00:00,30),|   (2017-10-09 00:00:00,90),|
|           |   (2017-06-14 00:00:00,30),|   (2018-07-11 00:00:00,90),|
|           |   (2017-03-16 00:00:00,30),|   (2017-04-16 00:00:00,90),|
|           |   (2017-08-14 00:00:00,30),|   (2018-09-16 00:00:00,90),|
|           |   (2017-09-12 00:00:00,30),|   (2018-04-14 00:00:00,90),|
|           |   (2017-01-21 00:00:00,30),|   (2018-01-12 00:00:00,90),|
|           |   (2018-08-14 00:00:00,30),|   (2017-07-15 00:00:00,90) |
|           |   (2018-08-23 00:00:00,30),| ]                          |
|           |   (2017-10-11 00:00:00,30),|                            |
|           |   (2017-12-12 00:00:00,30),|                            |
|           |   (2017-02-15 00:00:00,30),|                            |
|           |   (2018-04-12 00:00:00,30),|                            |
|           |   (2018-03-12 00:00:00,30),|                            |
|           |   (2018-05-15 00:00:00,30),|                            |
|           |   (2018-02-12 00:00:00,30),|                            |
|           |   (2018-06-14 00:00:00,30),|                            |
|           |   (2018-01-11 00:00:00,30),|                            |
|           |   (2017-07-20 00:00:00,30) |                            |
|           | ]                                                       |
+-----------+----------------------------+----------------------------+

如果您遇到枢轴性能问题，下面的方法是同一问题的另一种解决方案，尽管它允许您通过使用 for 循环将作业分为每个类别的阶段来获得更多控制。对于每次迭代，这会将类别_x 的新数据附加到 acc_df 中，该数据将保存累积结果。

schema = ArrayType( 
        StructType((  
            StructField("p_date", StringType(), False), 
            StructField("d_warranty", StringType(), False)  
        )) 
    )

    tuple_list_udf = udf(tuple_list, schema)

    buf_size = 5 # if you get OOM error decrease this to persist more often

    categories = df.select("category").distinct().collect()

    acc_df = spark.createDataFrame(sc.emptyRDD(), df.schema) # create an empty df which holds the accumulated results for each category

    for idx, c in enumerate(categories):
        col_name = c[0].replace(" ", "_") # spark complains for columns containing space
        cat_df = df.where(df["category"] == c[0]) \
                .groupBy("product_id") \
                .agg(
                    F.collect_list(F.col("purchase_date")).alias("p_date"), 
                    F.collect_list(F.col("days_warranty")).alias("d_warranty")) \
                .withColumn(col_name, tuple_list_udf(F.col("p_date"), F.col("d_warranty"))) \
                .drop("p_date", "d_warranty")

        if idx == 0:
            acc_df = cat_df
        else:
            acc_df = acc_df \
                .join(cat_df.alias("cat_df"), "product_id") \
                .drop(F.col("cat_df.product_id"))

        # you can persist here every buf_size iterations
        if idx + 1 % buf_size == 0:
            acc_df = acc_df.persist()

函数tuple_list 负责生成一个包含purchase_date 和days_warranty 列中的元组的列表。

def tuple_list(pdl, dwl):
    return list(zip(pdl, dwl))

其输出将是：

+-----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|product_id |CATEGORY_B                                                                                                                                                                                                                                         |CATEGORY_A                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
+-----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
|02147465400|[[2017-04-16 00:00:00, 90], [2018-09-16 00:00:00, 90], [2017-10-09 00:00:00, 90], [2018-01-12 00:00:00, 90], [2018-07-11 00:00:00, 90], [2017-01-21 00:00:00, 90], [2018-04-14 00:00:00, 90], [2017-01-05 00:00:00, 90], [2017-07-15 00:00:00, 90]]|[[2017-06-14 00:00:00, 30], [2018-08-14 00:00:00, 30], [2018-01-11 00:00:00, 30], [2018-04-12 00:00:00, 30], [2017-10-11 00:00:00, 30], [2017-05-16 00:00:00, 30], [2018-05-15 00:00:00, 30], [2017-04-15 00:00:00, 30], [2017-02-15 00:00:00, 30], [2018-02-12 00:00:00, 30], [2017-01-21 00:00:00, 30], [2018-07-11 00:00:00, 30], [2018-06-14 00:00:00, 30], [2017-03-16 00:00:00, 30], [2017-07-20 00:00:00, 30], [2018-08-23 00:00:00, 30], [2017-09-12 00:00:00, 30], [2018-03-12 00:00:00, 30], [2017-12-12 00:00:00, 30], [2017-08-14 00:00:00, 30], [2017-11-11 00:00:00, 30]]|
+-----------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

apachespark

DataFrame

PySpark

reduce

transpose

PySpark Dataframe 将两列转换为基于第三列值的新元组列的相关文章

如何使用 AWS Glue 开始处理从 Web 服务终端节点提取的数据？

更多源数据来自我需要定期轮询的 Web 服务端点获得数据后我可以使用 pyspark 执行传统的 ETL 并最终将数据写入 S3 和 Redshift 我不确定如何进行初始提取甚至不确定我应该在 AWS Glue 文档中查找什么内容
获取 int() 参数必须是字符串或数字，而不是“Column”- Apache Spark

如果我使用以下代码我会收到此异常 int argument must be a string or a number not Column df df withColumn FY F when df ID substr 5 2 isin
使用 stargazer 分析包含时间序列的数据帧

我有一个面板数据集共 10 个观测值和 3 个变量观测值 30 的数量 10 行国家地区 2 列迁移参数相应年份的 1 列可以这么说我的数据框由 3 个年度数据框组成我该如何申请观星者考虑到它是一个面板数据集所以最大 N
如何在 Databricks 中使用 OPTIMIZE ZORDER BY

我有两个数据框来自三角洲湖表它们通过 id 列进行左连接 sd1 sd2 sql select a columnA b columnB from sd1 a left outer join sd2 b on a id b id 问题是我
使用管道语法处理模型列表

我经常喜欢拟合和检查与 R 数据框中的两个变量相关的多个模型我可以使用如下语法来做到这一点 require tidyverse require broom models lt list hp exp cyl hp cyl map df m
计算 pyspark df 列中子字符串列表的出现次数

我想计算子字符串列表的出现次数并根据 pyspark df 中包含长字符串的列创建一个列 Input ID History 1 USA UK IND DEN MAL SWE AUS 2 USA UK PAK NOR 3 NOR NZE 4
根据 pyspark 中的条件从数据框中删除行

我有一个包含两列的数据框 col1 col2 22 12 2 1 2 1 5 52 1 2 62 9 77 33 3 我想创建一个新的数据框它只需要行 col1 的值 gt col2 的值就像注释一样col1 很长类型和col2 有双
根据 R 数据框中的名称对列进行平均

我想知道是否有一种有效的方法来获取每组的平均值类似命名的列谁的名字结尾为 1S and 2S ex ex1S ex2S at time 1并取每组的平均值类似命名的列谁的名字结尾为 1C or 2C ex ex1C ex2C at time
在 GPU 支持下对高维数据进行更快的 Kmeans 聚类

我们一直在使用 Kmeans 来对日志进行聚类典型的数据集有 10 mill 具有 100k 特征的样本为了找到最佳 k 我们并行运行多个 Kmeans 并选择轮廓得分最佳的一个在 90 的情况下我们最终得到的 k 介于 2 到 1
仅当某些值相等时，如何才能将一个文本文件中的值替换为另一个文本文件中的其他值？

我有一个名为finalscores txt我想创建一个 python 脚本它将打开它并从两个单独的列中读取值这是我的finalscores txt file Atom nVa predppm avgppm stdev delta QPr
Spark.sql.shuffle.partitions 的最佳值应该是多少，或者在使用 Spark SQL 时如何增加分区？

我实际上正在使用 Spark SQLhiveContext sql 它使用 group by 查询我遇到了 OOM 问题所以考虑增加价值spark sql shuffle partitions从默认的 200 到 1000 但这没有帮助
Spark scala 模拟 Spark.implicits 用于单元测试

当尝试使用 Spark 和 Scala 简化单元测试时我使用 scala test 和mockito scala 以及mockito Sugar 这只是让你做这样的事情 val sparkSessionMock mock SparkSes
Spark 在 WholeTextFiles 上创建的分区少于 minPartitions 参数

我有一个文件夹里面有 14 个文件我在一个集群上使用 10 个执行器运行 Spark Submit 该集群的资源管理器为 YARN 我创建了我的第一个 RDD 如下所示 JavaPairRDD
Scala 案例类忽略 Spark shell 中的导入

我希望这个问题有一个明显的答案我刚刚升级到 Spark v2 0 并且遇到了一个奇怪的问题火花外壳 Scala 2 11 版本如果我输入以下最小的 Scala import java sql Timestamp case class C
从 PySpark RDD 中的每个组中取出前 N 个元素（不使用 groupByKey）

我有一个如下所示的 RDD dataSource sc parallelize user1 3 blue user1 4 black user2 5 white user2 3 black user2 6 red user1 1 red 我
操作错误：(sqlite3.OperationalError) SQL 变量太多，同时将 SQL 与数据帧一起使用

我有一个熊猫数据框如下所示 activity User Id 0 VIEWED MOVIE 158d292ec18a49 1 VIEWED MOVIE 158d292ec18a49 2 VIEWED MOVIE 158d292ec18a4
一旦自定义 PySpark Transformer 成为安装的 ML Pipeline 中的一个阶段，如何为它设置参数？

我编写了一个自定义 ML PipelineEstimator and Transformer对于我自己的 Python 算法遵循所示的模式here https stackoverflow com a 37279526 1843329 然而
火花内存不足

我有一个文件夹里面有 150 G 的 txt 文件大约 700 个文件平均每个 200 MB 我使用 scala 来处理文件并最终计算一些汇总统计数据我认为有两种可能的方法可以做到这一点手动循环所有文件对每个文件进行计算并最终合
Spark：查找前 n 个值的高性能方法

我有一个很大的数据集我想找到具有 n 个最高值的行 id count id1 10 id2 15 id3 5 我能想到的唯一方法是使用row number没有分区就像 val window Window orderBy desc coun
如果两个阶段使用相同的 DataFrame，spark 是否会读取同一文件两次？

以下代码读取相同的 csv 两次即使只调用一个操作端到端可运行示例 import pandas as pd import numpy as np df1 pd DataFrame np arange 1 000 reshape 1 1

随机推荐

单页应用程序的搜索引擎索引

好吧我编写 Backbone js 应用程序已经一年多了我喜欢这个框架模型我已经学会了如何避免所有陷阱等但作为单页应用程序开发人员有一个领域我仍然很薄弱如何对面向公众的应用程序进行搜索引擎优化我正在开发一个博客项目我认为最简
使用 TypeScript 构建的 Angular 指令中的侧链接函数中的“this”未定义

我对 TypeScript 和 Angular 都是相对新手所以我可能在这里做了一些非常基本的错误我正在尝试创建一个抽象基类从中可以派生多个指令每个指令将实现一个自定义验证规则我的代码编译正常但在运行时失败特别是当它尝试调用
获取 varchar 一部分的最后一个单词（左/右）

获取 varchar 一部分的最后一个单词的正确方法是什么 DECLARE desc varchar 100 SET desc EXCHANGEUNIT P1i SILVERBLACK CYRILLIC SELECT RTRIM LEFT
在文件目录中重复创建 .htaccess 文件

我的共享主机中有大约 10 个站点最近我注意到创建了很多 htaccess 文件并限制了用户访问
使用 SwiftMailer 和 PHP 检索文件名以附加到电子邮件

我昨天问了这个问题并得到了建议并使用了它但由于某种原因它不起作用因此我需要检索用户从 HTML 表单上传到我的服务器的文件的名称我需要将此文件附加到由 PHP SwiftMailer 发送的电子邮件中这是我的代码文件上传部分
Wildfly -9 javax.mail 不工作

我正在使用 Wildfly 9 0 0 Final 和javax mail api在我的应用程序中我收到以下日志 2015 07 27 15 08 16 318 CONFIG javax mail default task 17 Prov
CLLocationManager 不要停止

亲爱的开发者朋友们我正在努力寻找解决 CLLocationManager 问题的方法我在我的应用程序中使用 CLLocationManager 实例如果用户选择设备上的主页按钮或终止应用程序我希望位置服务停止因此我打电话 se
将字符串解析为本地日期不使用所需的世纪

我正在使用这个 DateTimeFormatter DateTimeFormatter ofPattern ddMMYY 我想解析字符串150790我得到了这个错误 Unable to obtain LocalDate from Tempo
如何向每个 WCF 调用添加自定义 HTTP 标头？

我有一个托管在 Windows 服务中的 WCF 服务使用此服务的客户端每次调用服务方法时都必须传递一个标识符因为该标识符对于被调用方法应该执行的操作很重要我认为以某种方式将此标识符放入 WCF 标头信息中是个好主意如果这是个好主意
std::vector 内存在清除后是否被释放？

假设我有一个结构体 std vector 如果向量被clear 处理内存会发生什么 std vector
发布前离线保存数据

我想在发布之前将发布数据保存到用户电脑上以防万一我所在地区的互联网连接不好大多数时候用户撰写文章添加图像当单击提交按钮时他们的连接会由于某些问题而丢失那么 php ajax jquery 有没有办法我可以在打字时或发布之
WP7 IE - CSS 模式弹出窗口：点击/点击穿过覆盖 div 并触发应该不可见的链接

我正在制作一个针对移动设备的 HTML CSS 和基于 jQuery 的文件管理器其中一部分涉及使用基于 CSS 的模式对话框进行各种文件操作复制删除等我实现了这样的模式对话框行为 div div a href close a di
如果我不在 Android 中压缩纯文本文件，它是否会在 APK 中被压缩？

我正在与自己争论是否要压缩我的应用程序中包含的大约 1 5 MB 的纯文本文件我可以使用 zip 将其压缩到 400k 并在应用程序中解压缩但这只是另一件需要处理的事情如果我不这样做它会在 APK 中被压缩吗我可以把它放在一个特殊
Serde 使用远程对象的代理类型进行序列化

如何使用 Serde 为远程类型创建序列化器代理对象这是一个最小的例子操场 use serde 1 0 104 use serde json 1 0 48 struct Foo bar u8 impl Foo pub fn new gt
查找每个 id 的最大值，按日期分组[重复]

这个问题在这里已经有答案了可能的重复如何加快组内累加速度在下面的数据框中 id lt c 1 1 1 1 1 3 3 3 3 spent lt c 10 20 30 40 50 60 70 80 90 date lt c 11 11
注销页面在 Django 中不起作用

我正在尝试为 django 创建一个注销页面这是views py 文件 def index request if not request user is authenticated return redirect webapp login
如何对 OrderedDict 的 OrderedDict 进行排序？

我正在尝试对 OrderedDict 中的 OrderedDict 进行排序 depth 钥匙有什么解决方案可以对该 Dictionary 进行排序吗 OrderedDict 2 OrderedDict depth 0 height 51
C++11：在 std::array 上定义函数

std array采用两个模板参数 typename T the element type size t N the size of the array 我想定义一个函数它采用 std array 作为参数但仅适用于特定的 T 在本例中
Android - 应用程序未运行。错误：依赖项的 AAR 元数据中指定的 minCompileSdk 大于此模块的compileSdkVersion [重复]

这个问题在这里已经有答案了一周前该应用程序运行完美但突然没有更新任何内容该应用程序无法运行当我尝试运行该应用程序时出现此错误 minCompileSdk 31 中指定依赖项的 AAR 元数据 META INF com andr
PySpark Dataframe 将两列转换为基于第三列值的新元组列

正如主题所描述的我有一个 PySpark Dataframe 我需要将两列转换为一个新列它是基于第三列值的元组列表该演员阵容将减少或通过键值本例中为产品 id 展平数据框结果为一行每个键该数据框中有数亿行有 3700 万

PySpark Dataframe 将两列转换为基于第三列值的新元组列

PySpark Dataframe 将两列转换为基于第三列值的新元组列 的相关文章

随机推荐

热门标签

PySpark Dataframe 将两列转换为基于第三列值的新元组列的相关文章