在Pyspark中使用when条件的Groupby
在Pyspark中，可以使用when条件的Groupby来进行数据分组和聚合操作。when条件语句用于根据给定的条件对数据进行分类，根据条件的不同将数据分配到不同的组中。
下面是一种在Pyspark中使用when条件的Groupby的示例代码：
from pyspark.sql import SparkSession
from pyspark.sql.functions import when, count
# 创建SparkSession
spark = SparkSession.builder.appName("Groupby with When Condition in Pyspark").getOrCreate()
# 创建示例数据
data = [("Alice", 25, "F"),
("Bob", 30, "M"),
("Catherine", 35, "F"),
("David", 40, "M"),
("Emma", 45, "F")]
df = spark.createDataFrame(data, ["Name", "Age", "Gender"])
# 使用when条件的Groupby
grouped_df = df.groupBy(when(df.Gender == "F", "Female").otherwise("Male").alias("Gender")).agg(count("Name").alias("Count"))
# 显示结果
grouped_df.show()上述代码中，我们使用了when条件语句来根据"Gender"列的值进行分组，当"Gender"为"F"时，分配到"Female"组，否则分配到"Male"组。然后，我们使用agg函数计算每个组中的"Name"数量，并将结果保存到"Count"列中。最后，使用show方法显示结果。
这种方法适用于对数据集进行分类和聚合的场景，例如根据不同的条件对客户进行分组，并计算每个组的交易总额、平均年龄等。
在腾讯云的产品中，可以使用腾讯云的弹性MapReduce（EMR）服务来处理大数据和分布式计算任务。EMR提供了丰富的工具和资源，可以方便地在云端进行大规模数据处理和分析。您可以通过以下链接了解更多关于腾讯云EMR的信息：腾讯云EMR
注意：以上答案仅为示例，具体的答案可能因实际情况而有所不同。
相关·内容
1回答
我正在研究 stuck 方法，并坚持了如何实现以下结果。
星星之火sql (这很有效)
q = """
select breed,
avg(weight) as avg_wt,
avg(weight) filter (where age > 1) avg_wt_age_gt1
from cats
group by breed
order by breed
"""
spark.sql(q).show()
问:如何使用方法获得相同的结果？
我的尝试
(sdf.groupBy("breed").
浏览 2提问于2020-07-23得票数 0
回答已采纳
通常，当我必须进行聚合时，我使用类似于PySpark中的以下代码：
import pyspark.sql.functions as f
df_aggregate = df.groupBy('id')\
.agg(f.mean('value_col').alias('value_col_mean'))
现在，我实际上想要计算dataframe df的多个子集上的平均值(例如，在不同的时间窗口上，例如去年的平均值，过去两年的平均值，等等)。我知道我可以对每个子集执行df.filter(f.col(filter_co
浏览 4提问于2021-03-18得票数 1
回答已采纳
2回答
我遇到过这样的情况:我的列中可能有空值，需要在一个组中求和。 如果我在组中遇到null，我希望该组的和为null。但是默认情况下，Pyspark似乎会忽略null行，并对其余非NULL值求和。 例如： ? dataframe = dataframe.groupBy('dataframe.product', 'dataframe.price') \
.agg(f.sum('price')) 预期输出为： ? 但我得到的是： ? 任何建议都会有帮助！ 提前感谢！
浏览 87提问于2021-01-18得票数 1
回答已采纳
2回答
要Py的SQL查询(Spark)
、、、、
我有以下SQL查询，我想将其转换为pyspark。我想使用两个列pp和gender，并在pyspark中执行以下操作 %sql
SELECT pp
, SUM(CASE WHEN Gender = 'M'
THEN 1.0 ELSE 0.0 END) /
COUNT(1) AS gender_score
, count(1) AS total
FROM df
WHERE gender in ('M', 'F')
GROUP BY pp
HAVING
浏览 12提问于2020-12-16得票数 0
我有以下代码： from pyspark.sql import functions as sf
data = [{"Category": 'Category A', "ID": 1, "Value": 12.40},
{"Category": 'Category B', "ID": 2, "Value": 30.10},
{"Category": 'Category C', "ID": 3, "
浏览 16提问于2020-04-29得票数 0
我需要使用Pivot来转换表。我已经在pyspark sql中尝试过了，但遇到了问题。
Store Week xCount
------- ---- ------
102 1 96
101 1 138
105 1 37
109 1 59
101 2 282
102 2 212
105 2 78
109 2 97
105 3
浏览 2提问于2017-12-28得票数 3
回答已采纳
我无法对数据帧使用筛选器。我一直收到错误“TypeError(”条件应该是字符串或列“)” 我已经尝试更改滤镜以使用col对象。尽管如此，它还是不起作用。 path = 'dbfs:/FileStore/tables/TravelData.txt'
data = spark.read.text(path)
from pyspark.sql.types import StructType, StructField, IntegerType , StringType, DoubleType
schema = StructType([
StructField("from
浏览 21提问于2019-02-02得票数 0
回答已采纳
我正在玩Apache，并遇到了以下情况。我有一个名为“数据”的星火数据，类似于这种格式
Name Grade Count
X A 10
Y A 8
X B 2
Y B 4
现在，我想用‘groupBy’来表示这个数据，同时计算A级和B级之间的比率，例如X，它是10/2=5。
Name Ratio
X 5
Y 4
PySpark版本1.3.0
浏览 7提问于2016-01-07得票数 0
回答已采纳
1回答
将以下内容视为数据帧
a b c d e
africa 123 1 10 121.2
africa 123 1 10 321.98
africa 123 2 12 43.92
africa 124 2 12 43.92
usa 121 1 12 825.32
usa 121 1 12 89.78
usa 123 2 10 32.24
usa 123 5 21 43.92
canada 132 2 13 63.21
canada 132 2 13 89.23
can
浏览 2提问于2018-11-24得票数 1
4回答
我有几张桌子：
df1 df2
+---+------------+ +---+---------+
| id| many_cols| | id|criterion|
+---+------------+ +---+---------+
| 1|lots_of_data| | 1| false|
| 2|lots_of_data| | 1| true|
| 3|lots_of_data| | 1| true|
+---+------------+ | 3| false|
浏览 3提问于2021-03-01得票数 2
回答已采纳
1回答
考虑到以下数据
from pyspark.sql import functions as F
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
data = [
('a', 5, 'voice'),
('a', 8, 'sim'),
('a', 7, 'sim'),
('b', 1, 'handset'),
(
浏览 3提问于2021-07-15得票数 0
回答已采纳
1回答
我有一个测试火花环境(单节点)运行在AWS上。我在PySpark shell中执行了很少的临时查询，一切都如愿以偿，但是，当我使用spark submit运行应用程序时，会出现错误。
以下是代码：
from __future__ import print_function
from pyspark import SparkContext, SparkConf
from pyspark.sql.session import SparkSession
from pyspark.sql import SQLContext as sql
conf = SparkConf().setAppName(&
浏览 0提问于2017-11-10得票数 0
回答已采纳
我试图根据pyspark数据格式中的ID列对列进行分组，并根据另一列的值对列进行求和。
为了举例说明，请考虑以下虚拟数据格式：
+-----+-------+---------+
| ID| type| amount|
+-----+-------+---------+
| 1| a| 55|
| 2| b| 1455|
| 2| a| 20|
| 2| b| 100|
| 3| null| 230|
+-----+-------+---------+
我想
浏览 1提问于2019-06-13得票数 0
回答已采纳
我已将csv文件读入pyspark dataframe。现在，如果我在when()子句中应用条件，那么当条件在runtime之前给出时，它会很好地工作。
import pandas as pd
from pyspark import SparkContext
from pyspark.sql import SQLContext
from pyspark.sql import functions
from pyspark.sql.functions import col
sc = SparkContext('local', 'example')
sql_sc =
浏览 0提问于2019-11-08得票数 0
回答已采纳
我正在使用pyspark来读取一个数据，当列的值“is以外”==1的值时，它计算了"montant“的和。
但我知道这个错误：
文件""，第5行时(col(“isFraud”) =1，sum("montant")) ^ SyntaxError:关键字不能是表达式
在这里，代码：
CNP_df_fraude= (tx_wd_df
#.filter("isFraude =='1'").filter("POS_Card_Presence =='CardNotPresent&#
浏览 4提问于2019-10-04得票数 0
1回答
我有这样的数据： from pyspark.context import SparkContext
from pyspark.sql.session import SparkSession
sc = SparkContext.getOrCreate()
spark = SparkSession(sc)
from pyspark.sql.functions import substring, length
dept = [("A",1,"2020-11-07 23:19:12",50), ("A",1,"2020-11-07 23:19:
浏览 23提问于2021-01-18得票数 0
回答已采纳
pySpark中的新手提出了一个简单的问题:我有一个df，我想要进行一个条件累加，如果分母与0不同，则返回聚合结果。
我的试探性产生了一个错误：
groupBy=["K"]
exprs=[(sum("A")+(sum("B"))/sum("C") if sum("C")!=0 else 0 ]
grouped_df=new_df.groupby(*groupBy).agg(*exprs)
有什么暗示吗？
谢谢
浏览 3提问于2017-09-19得票数 2
回答已采纳
我有一个PySpark数据表，如下所示
shouldMerge | number
true | 1
true | 1
true | 2
false | 3
false | 1
我希望将所有列与shouldMerge合并为true，并将数字相加。
所以最后的输出看起来就像
shouldMerge | number
true | 4
false | 3
false | 1
如何选择所有具有shouldMerge == true的行，将数字相加，并在PySpark中生成一个新行？
编辑:替代的、
浏览 1提问于2019-02-21得票数 0
回答已采纳
2回答
无法替换空值的星火数据帧
、、、
下面的代码工作得很好，但是如果任何一个字段都是NULL ( SAL1, SAL2, SAL3, SAL4, SAL5列中的5列)，则相应的TOTAL_SALARY将以NULL的形式出现。看起来有些空条件或火花udfs需要创建，请您在这方面提供帮助。
投入：
NO NAME ADDR SAL1 SAL2 SAL3 SAL4 SAL5
1 ABC IND 100 200 300 null 400
2 XYZ USA 200 333 209 232 444
第二个记录的和很好，但是在第一个记录中，由于SAL4中的null，输出也是null。
from pyspark.she
浏览 1提问于2018-10-15得票数 0
回答已采纳
我在pyspark中有一个数据框架，如下所示
df = spark.createDataFrame(
[
("ABC",1809,'100.00'),
("ABC",851,'89.00'),
("ABC",852, '10.00'),
("BBC",1810,'10.00'),
("BBC",951,'8.90'),
("BBC",852, '1.00'),
("XXX",1810,&#
浏览 10提问于2022-03-15得票数 0
回答已采纳
1回答
我正在尝试将hql脚本转换为pyspark。我正在努力在groupby子句之后的聚合中实现case when语句的总和。例如：
dataframe1 = dataframe0.groupby(col0).agg(
SUM(f.when((col1 == 'ABC' | col2 == 'XYZ'), 1).otherwise(0)))
在pyspark中是可能的吗？我在执行这样的语句时遇到错误。谢谢
浏览 0提问于2016-11-23得票数 1
回答已采纳
1回答
我正在使用pyspark dataframe，我想添加一个列，看看单词是名词、动词、ADJ、ADV、ADP、PROPN，使用的是pyspark还是nltk.pos_tag
这是pyspark表。
------------------------
| event_dt | words |
------------------------
| 2020-09-02 | mifi |
| 2020-09-02 | hotspot |
| 2020-09-03 | service |
| 2020-09-03 | word |
| 2020-09-03 | plan |
这两
浏览 5提问于2020-11-03得票数 0
我有一个usecase，其中我想生成新的列值，请考虑如下所示的pyspark数据框架：
User | Product | Event | ...
----------------------------------------
ram apple viewed
ram apple carted
bill samsung viewed
bill samsung carted
bill samsung bought
我希望使用PySpark处理上面的数据
浏览 8提问于2020-10-23得票数 1
回答已采纳
假设您使用一个精确的模式创建了一个火花DataFrame：
import pyspark.sql.functions as sf
from pyspark.sql.types import *
dfschema = StructType([
StructField("_1", ArrayType(IntegerType())),
StructField("_2", ArrayType(IntegerType())),
])
df = spark.createDataFrame([[[1, 2, 5], [13, 74, 1]],
浏览 11提问于2022-11-29得票数 0
回答已采纳
我有下表：
df = spark.createDataFrame([(2,'john',1),
(2,'john',1),
(3,'pete',8),
(3,'pete',8),
(5,'steve',9)],
['id',&
浏览 1提问于2018-06-19得票数 1
回答已采纳
我是PySpark的新手，但我成功地完成了下面的工作。
不过，我还有两个需求，这两个需求都可以通过SQL中的case语句来实现。
我试过以下几种方法：
sqlfunc\
.when((df5.time_minute > 0) &(df5.time_minute < 16) , “Q1” )\
.when((df5.time_minute > 15) &(df5.time_minute < 31) , “Q2” )\
.when((df5.time_minute > 30) &(df5.time_minute < 46) , “Q3”
浏览 1提问于2018-09-21得票数 0
回答已采纳
1回答
我正在将Tableau中的一系列操作“翻译”成pyspark脚本。有一列我在复制时遇到了很大的麻烦，这里是代码的Tableu版本： min(
min(
IF[COLUMN_A] > 0 THEN [COLUMN_B] ELSE 9999 END,
IF[COLUMN_C] > 0 THEN [COLUMN_D] ELSE 99999 END),
min(
IF[COLUMN_E] > 0 THEN [COLUMN_F] ELSE 99999 END,
IF[COLUMN_G]
浏览 19提问于2021-07-15得票数 1
回答已采纳
1回答
我在pyspark数据帧中做一些聚合。我需要检查PAYMNT_STATUS列，在此基础上，我需要获取不同列的sum，并需要在新列中分配。另外，我对不同的状态进行计数，并将其分配到一个新列中。 我编写的pyspark查询如下所示： result=df1.groupby(countrry,year).withColumn('Amt1',sf.when(sf.col('PAYMNT_STATUS')==='A1',
sf.sum('Amt1_RS'))).withColumn("Amt2",sf.when(
浏览 30提问于2020-10-21得票数 0
我想得到情绪的比率，为此，我需要计算出每个话题有多少个正数和多少个否定词，然后除以每个主题的记录总数。
假设我有这个数据集：
+-----+---------+
|topic|sentiment|
+-----+---------+
|Chair| positive|
|Table| negative|
|Chair| negative|
|Chair| negative|
|Table| positive|
|Table| positive|
|Table| positive|
+-----+---------+
在这种情况下，我可以给出一个-1对‘负’和1到‘正’的值，那么这个比率在表(n
浏览 2提问于2022-04-02得票数 1
回答已采纳
这是SQL中的逻辑：
coalesce(if effc_dt <= tran_dt select(max of effc_dt) , if effc_dt >= tran_dt select (min of effc_dt))
我想在Pyspark中使用类似的逻辑，当effc date小于tran date时，它将选择最接近tran date的effc date，如果不存在较小的日期，它将检查更大的日期，并选择最接近tran date的effc date。
输入数据帧：
|id|tran_date |effc_date |
|--|-----------|-----------
浏览 8提问于2021-03-04得票数 1
回答已采纳
我正在和Spark和PySpark一起工作。我正在尝试实现与以下伪代码等效的结果：
df = df.withColumn('new_column',
IF fruit1 == fruit2 THEN 1, ELSE 0. IF fruit1 IS NULL OR fruit2 IS NULL 3.)
我正在尝试用PySpark来做这件事，但是我不确定它的语法。有什么建议吗？我查看了expr()，但无法让它正常工作。
请注意，df是一个pyspark.sql.dataframe.DataFrame。
浏览 0提问于2016-10-21得票数 57
回答已采纳
1回答
我在PySpark方面没有多少经验
我需要在检查有多少值大于某一阈值(绝对值)。我试过了，但没有用
n = lit(len(df.columns))
rank= (reduce(add, (1 for x in df.columns[1:] if abs(col(x)) > threshold))).alias(rank)
浏览 4提问于2018-04-29得票数 1
回答已采纳
1回答
我试图在PySpark数据帧上执行条件聚合。
我尝试了sum/avg，这似乎是正确的，但不知何故，计数给出了错误的结果。
from pyspark.sql import functions as F
df = spark.createDataFrame([('a', '1', 2502, 332),
('b', '1', 2328, 56),
('a', '1', 21, 78
浏览 0提问于2019-08-28得票数 0
回答已采纳
我正在解决这个例子：
在这里，他们使用Oracle或postgres命令nth_value来获得答案，但这不是在由pyspark使用的Hive中实现的，我想知道如何在pyspark中获得相同的结果。
postgres sql代码
所有大于4的重量被指定为第4最小的重量前三最轻的重量被赋值99.9
select name, weight,
coalesce(nth_value(weight, 4) over (order by weight), 99.9) as imagined_weight
from cats
order by weight
问:如何使用火花放电获得后续结果？
name
浏览 5提问于2020-07-21得票数 1
回答已采纳
我有一个spark数据帧，看起来像这样： +-------+-----------------------------+
|user_id| profile_features |
+-------+-----------------------------+
| 100 | [0.0, 0.33.., 0.66.., ...|
| 101 | [0.42..,0.15.., 0.57.., ...|
| 102 | [0.33..,0.0, 0.25.., ...|
| 103 | [0.15..,0.33.., 0.15.., ...
浏览 13提问于2020-08-24得票数 0
1回答
我想从分组的Spark数据帧中计算一列中有多少条记录是真的，但我不知道如何在python中做到这一点。例如，我有一个包含region、salary和IsUnemployed列的数据，其中IsUnemployed是一个布尔值。我想看看每个地区有多少失业人员。我知道我们可以先做一个filter，然后再做groupby，但是我想同时生成两个聚合，如下所示
from pyspark.sql import functions as F
data.groupby("Region").agg(F.avg("Salary"), F.count("IsUnemploy
浏览 0提问于2016-02-19得票数 17
回答已采纳
1回答
我在pyspark中有一个类似下面的数据框架。
+---+-------------+----+
| id| device| val|
+---+-------------+----+
| 3| mac pro| 1|
| 1| iphone| 2|
| 1|android phone| 2|
| 1| windows pc| 2|
| 1| spy camera| 2|
| 2| spy camera| 3|
| 2| iphone| 3|
| 3| spy came
浏览 34提问于2018-06-06得票数 0
回答已采纳
我在PySpark中有以下DataFrame： +----------+------------------------+
| id| codes_list|
+----------+------------------------+
| FF10| [[1049, 1683], [108]]|
| AB36| [[1507], [1005]]|
| 8266|[[1049], [1049], [1049]]|
+----------+------------------------+ 这是一个模式： r
浏览 28提问于2020-01-07得票数 2
1回答
我从一个csv文件开始，转换为下面的dataframe：
继续进行数据访问并使用PYSPARK，我需要在sensorID列中找到至少有两条满足条件的记录(PM10 > 50)的值。
然后，我需要有一个值为sensorID的输出和满足条件多少次的count。
输出应该是：sensorID: s1；2 (PM10>50计数)
我试过：
rdd.select("sensorID").where(col("PM10") > 50).count().show()
这给了我一个错误。我尝试了不使用.show()，但我不能只选择至少有两条记录的值(
浏览 1提问于2022-01-13得票数 0
回答已采纳
1回答
我正在尝试一些与pyspark相关的实验，在连接到AWS EMR实例的jupyter笔记本上。我有一个spark dataframe，它从s3读取数据，然后过滤掉一些东西。使用df1.printSchema()输出打印模式，如下所示： root
|-- idvalue: string (nullable = true)
|-- locationaccuracyhorizontal: float (nullable = true)
|-- hour: integer (nullable = true)
|-- day: integer (nullable = true)
|-- da
浏览 39提问于2019-05-09得票数 2
回答已采纳
from pyspark.sql import Row, functions as F
row = Row("UK_1","UK_2","Date","Cat")
agg = ''
agg = 'Cat'
tdf = (sc.parallelize
([
row(1,1,'12/10/2016',"A"),
row(1,2,None,'A'),
row(2,1,'14/10/2016
浏览 2提问于2017-10-18得票数 0
回答已采纳
1回答
将重复值转换为两个不同的列
、、、、
我将以下数据导入到PySpark数据框架中：
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSession.builder.master("local[4]").appName("pyspark").getOrCreate()
df = spark.read.csv("example.csv")
df.show()
#+--+------+
#|id|ans
浏览 10提问于2022-05-05得票数 1
回答已采纳
2回答
有没有一种方法可以比较PySpark中两个double类型的值，并具有指定的误差范围？Essential类似于this post，但在PySpark中。 类似于： df=#some dataframe with 2 columns RESULT1 and RESULT2
df=withColumn('compare', when(col('RESULT1')==col('RESULT2') +/- 0.05*col('RESULT2'), lit("match")).otherwise(lit("no
浏览 20提问于2021-02-08得票数 0
回答已采纳
1回答
我有以下SQL语句：
freecourse_info_step_8 as (
-- How many questions answered correct in that
select *,
count(question_number) FILTER (WHERE answered = true) over(partition by hacker_rank_id, freecourse_version, question_block, freecourse_users_id) as answered_correct_in_block
from freecourse_info_step
浏览 0提问于2020-08-12得票数 0
我在PySpark中工作，并将一个事务表作为Spark DataFrame导入，如下所示： User_ID Date Product_Name
-------- ------ -------------
A 2019-11-30. Product 1
B 2019-10-20 Product 2
C 2019-10-01 Product 1
A 2019-12-01 Product 1 我正在尝试做的是创建一个结果表，对于每个唯一的User_I
浏览 11提问于2020-10-03得票数 1
回答已采纳
1回答
当有存储在列中的数组时，我还不知道如何使用Pyspark，并在试图映射基于2 PySpark数据仓库的列时寻求一些帮助，其中一个是引用df。
参考书Dataframe (每个分组的数目各不相同)：
| Group | Subgroup | Size | Type |
| ---- | -------- | ------------------| --------------- |
|A | A1 |['Small','Medium'] | ['A','B'
浏览 4提问于2021-02-26得票数 0
回答已采纳
1回答
我尝试在PySpark中填充各种缺失的分类数据。我需要将所有的000和null映射为'UNDEFINED‘，但它将所有其他数据项转换为null。下面是我的输入 lastvalue_month.select('province').show(50) +--------------------+
| province|
+--------------------+
| DKI JAKARTA|
| DKI JAKARTA|
| JAWA BARAT|
| JAWA BARAT|
|
浏览 15提问于2021-09-06得票数 0
回答已采纳
可以看到，我们有两个数据帧currdf和hist_df，逻辑是在代码中定义的，预期的输出是
因此，在预期的输出中，我们可以观察到每次获得yellow marked data frame的迭代，总是有一行或没有行，我们需要在最终的数据帧后面附加行，如果数据帧(Tmp)是空的，那么只有pcode会存储为实际的，rest列将是空的。
复制代码
import pyspark
import os
import warnings
warnings.filterwarnings('ignore')
import pyspark.sql.functions as sf
fr
浏览 4提问于2021-06-14得票数 2
回答已采纳
它可以在没有聚合或计数的情况下遍历Pyspark groupBy数据帧吗？ 例如Pandas中的代码： for i, d in df2:
mycode ....
^^ if using pandas ^^
Is there a difference in how to iterate groupby in Pyspark or have to use aggregation and count?
浏览 54提问于2020-01-07得票数 3
回答已采纳
2回答
我想知道是否有一种方法可以在火花公子中使用.any()？
我有下面的Python代码，本质上是搜索子集dataframe中感兴趣的特定列，如果其中任何列包含"AD"，我们不想处理它们。
下面是Python中的代码：
index_list = [
df.query("id == @id").index
for trial in unique_trial_id_list
if ~(df.query("id == @trial")["unit"].str.upper().str.contains("A
浏览 7提问于2021-03-09得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
