到 Pyspark 的 SQL 查询）答案

【问题标题】：SQL Query to Py(spark)到 Pyspark 的 SQL 查询）
【发布时间】：2021-03-27 00:02:54
【问题描述】：

我有以下要转换为 pyspark 的 SQL 查询。我想取两列 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 count(1) > 100

我试过了

from pyspark.sql.functions import expr 
test_df = df.withColumn("gender_score", expr("CASE WHEN Gender == 'M' THEN  1 WHEN Gender == 'F' THEN  0 ELSE 'other' END AS male_score"))

但我不知道如何在 pyspark 中仅选择 2 列并将上述 SQL 查询复制到 pyspark

提前致谢

【问题讨论】：

标签： sql apache-spark pyspark count aggregation

【解决方案1】：

如果你想要一个 python 解决方案：

import pyspark.sql.functions as F

def df_func(df, grp):
    test_df = df.filter( 
        F.col('gender').isin(['M','F'])
        ).groupBy(grp).agg(
            F.mean(
                F.when(F.col('Gender') == F.lit('M'), 1.0).otherwise(0.0)
            ).alias('gender_score'),
            F.count(F.lit(1)).alias('total')
        ).filter('total > 100')
    return test_df

result = df_func(df, 'pp')

您可以通过创建数据框的临时视图来使用 Spark SQL：

df.createOrReplaceTempView('df')
query = """
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 count(1) > 100
"""
test_df = spark.sql(query)

【讨论】：

非常感谢。但想法是让它 py(spark) 不触发 SQL
@JamesTaylor 使用 pyspark 解决方案查看已编辑的答案。
对不起，如果它太多了。可以将其转换为我可以使用的功能吗？唯一会改变的参数是 groupBy :)
我添加了这个/count(lit(1)).alias('gender_score') ，但我得到了奇怪的列名
@JamesTaylor 如果要计算平均值，请使用 F.mean，请参阅编辑后的答案。

【解决方案2】：

groupBy agg() 函数采用多个表达式。所以你可以做一个 sum/count 表达式。

val df = spark.sql(""" with t1(
select 10 pp, 'M' gender union all
select 10 pp, 'F' gender union all
select 10 pp, 'M' gender union all
select 40 pp, 'F' gender 
) select * from t1
""")

df.show(false)

+---+------+
|pp |gender|
+---+------+
|10 |M     |
|10 |F     |
|10 |M     |
|40 |F     |
+---+------+

df.groupBy("pp").agg (
    (sum(expr("CASE WHEN Gender = 'M' THEN  1 ELSE 0 END "))/count(lit(1)) ).alias("gender_score"),
    count(lit(1)).alias("total")
).show(false)

+---+------------------+-----+
|pp |gender_score      |total|
+---+------------------+-----+
|40 |0.0               |1    |
|10 |0.6666666666666666|3    |
+---+------------------+-----+

之后，您可以使用另一个过滤器来实现“拥有”。

【讨论】：