Validate Multiple email address columns and concate both columns value into 1 column value delimited by pipe using pyspark dataframe

user_16396136 0
Viewed 277

I have two email address in pyspark dataframe and need to validate both the email address columns value and then merge into 1 column value using pipe as delimiter.

Primary_Email_Address       Alternate_Email_Address      Output(email_address)
navg@gmail.com                nvgng@gm@.com              navg@gmail.com
ggg@gmail.com                 nnnn@gmail.com             ggg@gmail.com|nnnn@gmail.com
nave@gmail.com|gg@gmail.com   ndg@gmail.com          nave@gmail.com|gg@gmail.com|ndg@gmail.com

I need to pass columns dynamically from spark dataframe csv file and handle it.

I have written below code but need some modification.

from pyspark.sql.session import SparkSession
from pyspark.sql.functions import concat, lit, col, trim, when
from pyspark.sql import functions as F

spark = SparkSession.builder.getOrCreate()
regex = '^\w+([\.-]?\w+)*@\w+([\.-]?\w+)*(\.\w{2,3})+$'

df = spark.read.option("header",True).csv("file_name.csv")

# select specific columns

df1 = df.select("primary_email_addr","alt_email_addr")

# concat the columns

df_email = df1.withColumn('email_addr',concat(df1['primary_email_addr'],lit("|"),df1['alt_email_addr'])))

df_email_1 = df_email.drop("primary_email_addr","alt_email_addr")

df_email_2 = df_email_1.withColumn("email_addr_status",when(F.col("email_addr").rlike("^(([a-zA-Z0-9_.+-]+)+@.+\\.[A-Za-z]+)(| ([a-zA-Z0-9_.+-]+)+@.+\\.[A-Za-z]+)*$"),"Valid").otherwise("Invalid"))

df_email_2.show(truncate=False)
pyspark apache-spark-sql apache-spark python
2 Answers

Here another approach using regexp_extract_all and array_join with Spark >= 3.1.0:

from pyspark.sql.functions import expr

cols = ['Primary_Email_Address', 'Alternate_Email_Address']
data = [("navg@gmail.com", "nvgng@gm@.com"),
        ("ggg@gmail.com", "nnnn@gmail.com")]

df = spark.createDataFrame(data, cols)

concat_expr = expr("concat_ws(',', Primary_Email_Address, Alternate_Email_Address)") 
extract_expr = expr("regexp_extract_all(concat, '(\\\w+([\\\.-]?\\\w+)*@\\\w+([\\\.-]?\\\w+)*(\\\.\\\w{2,3})+)', 0)")
join_expr = expr("array_join(extract, '|')")

df.withColumn("concat", concat_expr) \
  .withColumn("extract", extract_expr) \
  .withColumn("output", join_expr) \
  .select("Primary_Email_Address", "Alternate_Email_Address", "output") 

# +---------------------+-----------------------+----------------------------+
# |Primary_Email_Address|Alternate_Email_Address|output                      |
# +---------------------+-----------------------+----------------------------+
# |navg@gmail.com       |nvgng@gm@.com          |navg@gmail.com              |
# |ggg@gmail.com        |nnnn@gmail.com         |ggg@gmail.com|nnnn@gmail.com|
# +---------------------+-----------------------+----------------------------+

Logic:

  1. Use concat_ws to concatanate emails.
  2. Call regexp_extract_all to extract the array of valid emails.
  3. Call array_join to produce s string from the previous array with items seperated by |.
user_7503760

I am not sure if you looking for below solution or not -

  1. compare both the column
  2. If there is a mismatch - identify it
  3. Based on the identification, merge again

Modified my answer - Used from validate_email import validate_email module to validate the email

DF

    from validate_email import validate_email

df = spark.createDataFrame([("navg@gmail.com","nvgng@gm@.com "),("ggg@gmail.com","nnnn@gmail.com")],["Primary_Email_Address","Alternate_Email_Address"])

valid_email_udf = udf(lambda e: validate_email(e), T.BooleanType())
df = df.withColumn('is_valid_primary', valid_email_udf('Primary_Email_Address'))
df = df.withColumn('is_valid_alternate', valid_email_udf('Alternate_Email_Address'))
df.show()

Condition and final output

df = df.withColumn("cond_col", F.when(F.col("is_valid_primary") == F.col("is_valid_alternate"), True).otherwise(False))
df = df.withColumn("merged_email", F.when(F.col("cond_col") == True, F.concat_ws("|","Primary_Email_Address", "Alternate_Email_Address" )).otherwise(None))
df.show(truncate=False)

+---------------------+-----------------------+----------------+------------------+--------+----------------------------+
|Primary_Email_Address|Alternate_Email_Address|is_valid_primary|is_valid_alternate|cond_col|merged_email                |
+---------------------+-----------------------+----------------+------------------+--------+----------------------------+
|navg@gmail.com       |nvgng@gm@.com          |true            |false             |false   |null                        |
|ggg@gmail.com        |nnnn@gmail.com         |true            |true              |true    |ggg@gmail.com|nnnn@gmail.com|
+---------------------+-----------------------+----------------+------------------+--------+----------------------------+
user_51082200
Related