How to improve this application of survivorship and preferable do it without a UDF in Pyspark databricks?

Question

Hello, I am working on a project that involves collecting data from various sources. I have a scenarios where I have a final table of more than 800k records.

Since there are multiple sources there are certain sources which are preferred for a column and due to this I have a table which contains weightage given to a column from a certain source.( Please refer to the table below that will explain it better). Basically it is the mapping of weights (the lower the number, higher will be the priority) through which the column value is given precedence over the same column value from different source.

Sample weightage table:

|   columns   |. source1. |. source2. |. source3  |...| source10|
|-------------|-----------|-----------|-----------|---|---------|
|first_name   |     1     |2          |3          |...|10       |
|last_name.   |3          |1          |2          |...|9        |
|mobile       |2          |3          |5          |...|8        |
|email        |2          |3          |5          |...|8        |
|city         |3          |5          |1          |...|9        |

Now I have a created a look up table for the identification of source system based on numbers. Basically I have assigned a number to a source system. (PS: I need this table for other action items as well. Can't get away with this.) Here is a sample table:

|source system|ID|
|-------------|--|
|source1      |11|
|source3.     |12|
|source4.     |13|
|source5.     |14|
|source6.     |15|

Now, let's come to the final table. This is created post cleansing and has all the records. The table contains around 90 columns.

|first_name   |. last_name. |. mobile. |. email      |city.    |source system ID|
|-------------|-------------|----------|-------------|---------|----------------|
|john         |     doe     |12345667  |john@xyz.com |london   |12              |
|j            |             |12345667. |jack@abc.com |new york |13              |
|john         |d            |12345667. |john@abc.com |         |11              |
|jack         |dawn         |344567754 |jack@abc.com |amsterdam|11              |
|j            |dawn         |344567754 |jack@abc.com |zurich   |13              |

Now I need to merge the records (let's say on mobile or email) from different source systems. Basically each row is from a different source system. For this what I have to do this I'll gather the records based on mobile (and the later on email) then I'll check each column from weight table. Where ever I find the record with less weight it means it has more priority and I'll consider the value of that source system.

Finally after matching all these records I have to generate 1 single record. Now the problem is I have to do this on:

on source table alone and

also on the input data table (that is for upcoming data; so have to match the upcoming record from the existing records)

Here is how I am doing it for source table only scenario. Below is the excerpt of the UDF that I have created:


    def match_merge_surviver(col, col_input, inp_source_system_id, mobile):
     inp_weight = '999'
     temp = []
     first_name = {
          '17':1,
          '16':9,
          '14':8,
          '4':4,
          '13':3,
          '18':5,
          '19':2,
          '20':1,
          '21':1,
          '23':6,
          '11':999,
          '24':1
          }
      
      if(not col_input or col_input =='Not Provided' or col_input == 'Null'):
        return col_input
          
      if col =='first_name':
        temp={}
        try:
          if temp:
            if temp[mobile] > inp_weight:
              temp[mobile] = inp_weight
              return col_input
            else:
              pass
          else:
            inp_weight = first_name[str(inp_source_system_id)]
            temp[mobile] = inp_weight
            return col_input
        except KeyError:
                # Key is not present
                inp_weight = 999
                pass
    
       
       
    #Registering function
    sqlContext.udf.register("survivorship", survivorship)

Here is this Spark SQL query I am using for the application of UDF. Please note in this approach I have to run this UDF on each of 80+ columns. I am showing this only for first_name. Also, in where clause I have given mobile but it needs to run for thousands pf records.

```
%sql
select a.email, a.source_system_id, a.mobile, a.DOB, a.first_name,
match_merge_surviver('first_name',a.first_name, cast(a.source_system_id as 
string),a.mobile) 
as sur_first_name 
from final_table a
where a.source_system_id <> 0 
and a.source_system_id is not null 
and a.mobile = '12345667'

```

The output after above operation is:

|first_name   |. last_name. |. mobile. |. email     |city.   |source system ID|
|-------------|-------------|----------|------------|--------|----------------|
|john         |    doe      |12345667  |john@xyz.com|london  |12              |
|j            |             |12345667. |jack@abc.com|        |13              |
|john         |d            |12345667. |john@abc.com|brooklyn|11              |

Now it's not very effective since now I have to do a complex and expensive join using something like :

```
ROW_NUMBER() OVER (PARTITION BY mobile ORDER BY survived_first_name, 
modified_date desc )

```

The output should be like a 1 single record with all matching done:

|first_name   |. last_name. |. mobile. |. email      |city   |source system ID|
|-------------|-------------|----------|-------------|-------|----------------|
|john         |    doe      |12345667  |john@xyz.com |london |12              |

which again is not very effective and since I am already creating a UDF taking a performance hit.

Can you please help me improve this approach and if possible suggest a non UDF way. I'll really appreciate your help. Apologies for the long post but I needed to and tried to explain the whole process. Any further questions are highly appreciated.

CRAFTY DBA · Answer

Here is some sample code to play around with. I used the simple idea the source system ranked weight determines which row to pick. I would create a table out of the source to rank mapping and do all the work in Spark SQL.

Here is the code to create the people data.

# tuples - data + column
people_data = [ 
  ("john", "    doe ", "12345667", "john@xyz.com", "london", 12), 
  ("j", "", "12345667.", "jack@abc.com", "new york", 13), 
  ("john", "d", "12345667.", "john@abc.com", "", 11), 
  ("jack", "dawn", "344567754", "jack@abc.com", "amsterdam", 11), 
  ("j", "dawn", "344567754", "jack@abc.com", "zurich", 13) 
]
people_cols = ["first_name", "last_name", "mobile", "email", "city", "source_system_id"]

# create dataframe
df1 = spark.createDataFrame(data=people_data, schema = people_cols)

# create temp view
df1.createOrReplaceTempView("tmp_people_data")

Here is the code to create the ranking (weight) data.

# tuples - data + column
rank_data = [ 
  (4, 4),
  (11, 999),
  (13, 3),
  (14, 8),
  (16, 9),
  (17, 1),
  (18, 5),
  (19, 2),
  (20, 1),
  (21, 1),
  (23, 6),
  (24, 1)
]
rank_cols = ["source_system_id", "wgt_value"]

# create dataframe
df2 = spark.createDataFrame(data=rank_data, schema = rank_cols)

# create temp view
df2.createOrReplaceTempView("tmp_rank_data")

Last but not least, left join on source system, convert null weights to 999, create an row id based on partition and select the one that you want.

Here is the spark sql.

%sql
with cte1 as
(
select 
  p.*,
  nvl(r.wgt_value, 999) as wgt_value
from 
  tmp_people_data as p
left join
  tmp_rank_data as r
on 
  p.source_system_id = r.source_system_id
),
cte2 as
(
select 
  *,
  row_number() over (partition by mobile order by wgt_value desc) as rid
from cte1
)
select * from cte2 where rid = 1

This one gives us the rows that are ranked from the highest sources.

How to improve this application of survivorship and preferable do it without a UDF in Pyspark databricks?

1 Answers