以下 Databricks SQL 的等效 Py Spark 代码是什么

Question

我有下面的 Databricks SQL 代码（带有过滤条件），想要转换为 Py Spark 代码，但无法得到任何想法。 我搜索了谷歌，但只得到了 Py Spark 过滤器条件，但我没有看到任何适合我的用例的示例

    select
    account_doc_num,

    ROUND(coalesce(Sum(tbl.Amnt) FILTER(WHERE tbl.AmntType='type1' AND tbl.line_type='D'),0),2) AS Sls_Inv,
    ROUND(coalesce(Sum(tbl.Amnt) FILTER(WHERE tbl.AmntType='type1' AND tbl.line_type='GVAT'),0),2) AS GVAT,

    ROUND(coalesce(Sum(tbl.Amnt) FILTER(WHERE tbl.AmntType='type2' AND tbl.line_type='vat'),0),2) AS VAT,
    ROUND(coalesce(Sum(tbl.Amnt) FILTER(WHERE tbl.AmntType='type2' AND tbl.line_type='K'),0),2) AS Pur_Inv,

    ROUND((Sls_Inv + GVAT + VAT + Pur_Inv),2) AS Diff
    
    from
    DeltaTable as tbl
    group by account_doc_num

下面是PySpark代码，想要添加sum函数的过滤条件。 我想要的是具有 where 条件的 Databricks Filter 的等效 Py Spark 函数/代码

df_group=(
    df_vat.groupby("account_doc_num")
    .agg(
       sum("Amnt").alias("Sls_Inv"),
       sum("Amnt").alias("GVAT"),
       sum("Amnt").alias("VAT"),
       sum("Amnt").alias("Pur_Inv")
    )
)

非常感谢任何帮助或文档链接

Answer 1

您可以对 case/when 表达式进行求和，例如：

from pyspark.sql import functions as F

.agg(
       F.sum(F.when(F.col(AmntType)=='type1' AND F.col("line_type") =='D', F.col("Amnt")).otherwise(0)).alias("Sls_Inv"),

当条件为真时，这将计算 Amnt，否则该行提供 0。只要它们返回 0 并且其中没有聚合函数（例如，没有嵌套聚合），您就可以创建相当数量的表达式复杂性。

有多种方法可以在代码中构建它，谷歌搜索 sumif pyspark 和 Spark 组合。