How can I write conditional regex replace in PySpark?

Question

I'm trying to get an if else statement working in the form of when() and otherwise() function. I tried thinking of many ways but my problem is the when() statement needs a column. Is there a way to get the following code working in when() and otherwise() format?

get_data = spark.sql("SELECT STRING({}) AS {} FROM {} WHERE Mobile='{}'".format(dynamic_tag_mapping_column_name, match[0], dynamic_tag_mapping_table_name, mobile_numbers[mob])).collect()[0][0]
if get_data!='0':
     textList.append(campaign_segment_text.withColumn('CampaignSMSText', func.expr("regexp_replace(CampaignSMSText, '{}', {})".format(match[0], get_data)))
else:
     textList.append(campaign_segment_text.withColumn('CampaignSMSText', func.expr("regexp_replace(CampaignSMSText, '{}', {})".format(match[0], dynamic_default_value)))

What I'm trying to achieve is if I get the get_data>0 value from database it will take the get_data value but if get_data=0, it will assign a default value that is set. I tried setting the String value to Int that I'm getting from the DB but that won't compare because get_data isn't a column. Is there anyway to get this working?

mck · Accepted Answer · 2020-12-12T15:44:34.667

1

Try putting case when inside F.expr?

get_data = spark.sql("SELECT {} AS {} FROM {} WHERE Mobile='{}'".format(dynamic_tag_mapping_column_name, match[0], dynamic_tag_mapping_table_name, mobile_numbers[mob])).collect()[0][0]

textList.append(
    campaign_segment_text.withColumn(
        'CampaignSMSText',
        func.expr(
            "regexp_replace(CampaignSMSText, '{}', case when {} != 0 then {} else {} end)".format(
                match[0], get_data, get_data, dynamic_default_value
            )
        )
    )
)

edited Dec 12 '20 at 15:44

answered Dec 12 '20 at 15:25

mck

40,932
13
35
50

@VarunNagrare I found a missing close bracket in your code (and also my code), could you please try again? – mck Dec 12 '20 at 15:44
Okay I'll check it – Varun Nagrare Dec 12 '20 at 15:45
Yeah it worked! Thanks man. Your answer helped me a second time. :D – Varun Nagrare Dec 12 '20 at 15:49
I know UDF's are slow. I just wanted to practice since I don't know how UDF's work. – Varun Nagrare Dec 12 '20 at 15:53
yes you can use udf. the udf will take a string, pattern, and pattern to replace as input arguments, and return a replaced string – mck Dec 12 '20 at 15:57
I read somewhere that we can convert a Python function to Spark UDF. What steps would that need? – Varun Nagrare Dec 12 '20 at 15:59
see https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.functions.udf – mck Dec 12 '20 at 16:00

How can I write conditional regex replace in PySpark?

1 Answers1

Linked