(select sum(case faturado.shkzg

when 'S' then faturado.menge

else - faturado.menge

end)

from sapsr3.ekbe faturado where faturado.ebeln = b.ebeln and b.ebelp = faturado.ebelp and faturado.vgabe = '1' -- entradas

) as qtd_faturada

def qtd_faturada(col1,col2):

return (sum(when(col1=="S",col2)

.otherwise(-(col2))))

def with_qtd_faturada(df):

return df.withColumn("QTD_FATURADA",qtd_faturada(col("SHKZG"),col("MENGE")))

import pandas as pd
from pyspark.sql import functions as f


df = spark.createDataFrame(pd.DataFrame([
  {'SHKZG': 'S', 'MENGE': 1},
  {'SHKZG': 'S', 'MENGE': 1},
  {'SHKZG': 'N', 'MENGE': 1},
  {'SHKZG': 'N', 'MENGE': 1},
]))

def with_qtd_faturada(df, col1, col2):
  # create sum of column
  df_with_sum = df.select(f.sum(col1).alias('QTD_FATURADA'))

  # join with original df and flip sign based on col2's value
  return df.join(df_with_sum).withColumn(
    'QTD_FATURADA', f.when(col2=='S', f.col('QTD_FATURADA')).otherwise(-f.col('QTD_FATURADA'))
  )

with_qtd_faturada(df, f.col('MENGE'), f.col('SHKZG')).show()