2

I am trying to find a way on how I could run for loop to better optimize my script for my case statement.

The script shown below has no error however I feel this is too long-winded which can cause confusion during the next maintenance.

df = df.withColumn('Product', when(df.where('input_file_name LIKE "%CAD%"'), 'Cash and DUE').
                   when(df.where('input_file_name LIKE "%TP%"'), 'Trade Product').
                   when(df.where('input_file_name LIKE "%LNS%"'), 'Corp Loans').
                   when(df.where('input_file_name LIKE "%DBT%"'), 'Debt').
                   when(df.where('input_file_name LIKE "%CRD%"'), 'Retail Cards').
                   when(df.where('input_file_name LIKE "%MTG%"'), 'Mortage').
                   when(df.where('input_file_name LIKE "%OD%"'), 'Overdraft').
                   when(df.where('input_file_name LIKE "%PLN%"'), 'Retail Personal Loan').
                   when(df.where('input_file_name LIKE "%CLN%"'), 'CLN').
                   when(df.where('input_file_name LIKE "%CAT%"'), 'Custody and Trust').
                   when(df.where('input_file_name LIKE "%DEP%"'), 'Deposits').
                   when(df.where('input_file_name LIKE "%STZ%"'), 'Securitization').
                   when(df.where('input_file_name LIKE "%SECZ%"'), 'Security Securitization').
                   when(df.where('input_file_name LIKE "%SEC%"'), 'Securities').
                   when(df.where('input_file_name LIKE "%MTSZ%"'), 'Retail Mortage Securitization').
                   when(df.where('input_file_name LIKE "%PLSZ%"'), 'Retail Personal Loan Securitization').
                   when(df.where('input_file_name LIKE "%CCSZ%"'), 'Retail Cards Securitization').
                   when(df.where('input_file_name LIKE "%CMN%"'), 'Cash Management').
                   when(df.where('input_file_name LIKE "%OTC%"'), 'Over-the-counter').
                   when(df.where('input_file_name LIKE "%SFT%"'), 'Securities Financing Transactions').
                   when(df.where('input_file_name LIKE "%ETD%"'), 'Excahnge Traded Deriative').
                   when(df.where('input_file_name LIKE "%DEF%"'), 'Default Products').
                   when(df.where('input_file_name LIKE "%FFS%"'), 'Not Required').
                   when(df.where('input_file_name LIKE "%hdfs%"'), 'Not Required').
                   otherwise('feed_name'));

I have thought of running a loop, an example is shown below (scripts is not correct, it's for demo purposes)

product_code = ['%CAD%','%TP%','%LNS%','%DBT%','%CRD%','%MTG%','%OD%','%PLN%','%CLN%','%CAT%','%DEP%','%STZ%','%SECZ%','%SEC%','%MTSZ%','%PLSZ%','%CCSZ%','%CMN%','%OTC%','%SFT%','%ETD%','%DEF%','%FFS%','%hdfs%']
product_name = ['Cash and Due','Trade Product','Corp Loans','Debt','Retail Cards','Mortage','Overdraft','Retail Personal Loan','CLN','Custody and Trust','Deposits','Securitization','Securities Securitization','Securities','Retail Mortage Securitization','Retail Personal Loan Securitization','Retail Cards Securitization','Cash Management','Over-the-counter','Securities Finanacing Transactions','Exchange Traded Derivative','Default Products','Not Required','Not Required']
   
##Both product_code & product name have the same number of index

lastIndex = len(product_code)    
    for x in product_code:
       # Logic i thought df.withColumn('Product', when(df.where('input_file_name LIKE "%'product_code[x]'%"'), product_name[x])
       if(product_code[lastIndex]): 
      #otherwise('feed_name')

Would need some advice if running loop for case statement for when(df.where()).otherwise is possible in spark or there is another approach or use-case I can look up to

UPDATED

I have implemented with the method as adviced, query is returning correct on the condition set but I was wondering why its not returning the correct value on otherwise known as lit() in script below, instead its removing the row that does not meet the condition

Sample DF:
product_code = ['%CMN%','%TP%','%LNS%']
product_name = ['Cash and Due','Trade Product']
feed_name = ['farid','arshad','jimmy']   

df = spark.createDataFrame(
     list(zip(inp_file,feed_name)),
     ['input_file_name','feed_name']
)

+---------------+---------+
|input_file_name|feed_name|
+---------------+---------+
|sdasdasdasd    |bob      |
|_CMN_BD        |arshad   |
|_CMN_BD_WS     |jimmy    |
+---------------+---------+

product_code = ['%CAD_%','%TP%','%LNS%','%DBT%','%CRD%','%MTG%','%_OD_%','%PLN%','%CLN%','%CAT%','%DEP%','%STZ%','%SECZ%','%SEC%','%MTSZ%','%PLSZ%','%CCSZ%','%CMN%','%OTC%','%SFT%','%ETD%','%DEF%','%FFS%','%hdfs%']
product_name = ['Cash and Due','Trade Product','Corp Loans','Debt','Retail Cards','Mortage','Overdraft','Retail Personal Loan','CLN','Custody and Trust','Deposits','Securitization','Securities Securitization','Securities','Retail Mortage Securitization','Retail Personal Loan Securitization','Retail Cards Securitization','Cash Management','Over-the-counter','Securities Finanacing Transactions','Exchange Traded Derivative','Default Products','Not Required','Not Required']
   
## -- Create spark dataframe and with list tuple     
## -- Lit is used to add new column

product_ref_df = spark.createDataFrame(
     list(zip(product_code, product_name)),
     ["product_code", "product_name"]
)
    

def tempDF(df,targetField,columnTitle,condition,targetResult,show=False):
    product_ref_df = spark.createDataFrame(
         list(zip(condition,targetResult)),
         ["condition", "target_result"]
    )
    
    df.join(broadcast(product_ref_df), expr(""+targetField+" like condition")) \
    .withColumn(columnTitle, coalesce(col("target_result"), lit("feed_name"))) \
    .drop('condition','target_result') \
    .show()
    
    return df

product_ref_df = tempDF(df,'input_file_name','Product',product_code,product_name)

When script is triggered, there is no error and the result return as shown,

+---------------+---------+------------+
|input_file_name|feed_name|     Product|
+---------------+---------+------------+
|        _CMN_BD|   arshad|Cash and Due|
|     _CMN_BD_WS|    jimmy|Cash and Due|
+---------------+---------+------------+

Shouldnt the result return with the first row in it since we are not removing any row,

+---------------+---------+------------+
|input_file_name|feed_name|     Product|
+---------------+---------+------------+
|    sdasdasdasd|      bob|bob         |
|        _CMN_BD|    jimmy|Cash and Due|
|     _CMN_BD_WS|    jimmy|Cash and Due|
+---------------+---------+------------+
+---------------+---------+------------+

2 Answers 2

4

You could create a new DataFrame from those product names referential and join with the original df to get the product name :

from pyspark.sql.functions import expr, col, broadcast, coalesce

product_ref_df = spark.createDataFrame(
     list(zip(product_code, product_name)),
     ["product_code", "product_name"]
)

df.join(broadcast(product_ref_df), expr("input_file_name like product_code"), "left") \
  .withColumn("Product", coalesce(col("product_name"), col("Feed_name"))) \
  .drop("product_code", "product_name") \
  .show()

Or use functools.reduce to chain the case/when conditions like this:

import functools

from pyspark.sql.functions import lit, col, when

case_conditions = list(zip(product_code, product_name))

product_col = functools.reduce(
    lambda acc, x: acc.when(col(f"input_file_name").like(x[1]), lit(x[1])),
    case_conditions[1:],
    when(col("input_file_name").like(case_conditions[0][0]), lit(case_conditions[0][1]))
).otherwise(col("Feed_name"))

df.withColumn("Product", product_col).show()
2
  • a bit dfferent from pyspark.sql import functions as F product_col = functools.reduce( lambda acc, x: acc.when(F.col(f"input_file_name").like(x[1]), F.lit(x[1])), case_conditions, F, ).otherwise("Feed_name")
    – Steven
    Jan 19, 2021 at 9:25
  • i implemented the approach u have given, however, lit() otherwise is not returning any value , instead its removing the rows which like condition is not met. i have updated my question for your reference. Jan 22, 2021 at 5:12
3

You can use coalesce to combine all the when statements into one. coalesce will pick the first non-null column, and when only gives non-null if the condition matches; otherwise it gives null (without an otherwise condition).

product_code = ['%CAD%','%TP%','%LNS%','%DBT%','%CRD%','%MTG%','%OD%','%PLN%','%CLN%','%CAT%','%DEP%','%STZ%','%SECZ%','%SEC%','%MTSZ%','%PLSZ%','%CCSZ%','%CMN%','%OTC%','%SFT%','%ETD%','%DEF%','%FFS%','%hdfs%']
product_name = ['Cash and Due','Trade Product','Corp Loans','Debt','Retail Cards','Mortage','Overdraft','Retail Personal Loan','CLN','Custody and Trust','Deposits','Securitization','Securities Securitization','Securities','Retail Mortage Securitization','Retail Personal Loan Securitization','Retail Cards Securitization','Cash Management','Over-the-counter','Securities Finanacing Transactions','Exchange Traded Derivative','Default Products','Not Required','Not Required']

import pyspark.sql.functions as F

df2 = df.withColumn(
    'Product',
    F.coalesce(
        *[F.when(F.col('input_file_name').like(code), F.lit(name))
        for (code, name) in zip(product_code, product_name)]
    )
)
1
  • Could you please add links to important functions (coalesce in this case) if possible? You can do it using [function](link). It's easier to follow. Jan 19, 2021 at 8:21

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.