Notes

# fully converted to Splink 4.0, with all Java UDFs removed and replaced by native Spark SQL functions.
import time
import argparse
import sys
import gc
import os
import re
import boto3
import warnings

from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import (
col, concat_ws, countDistinct, collect_list, lit, split, explode,
row_number, desc, round, when
)
from pyspark.sql.types import *

from splink import Splink

# Suppress warnings
warnings.filterwarnings('ignore')

# Argument parser
def _parse_args(argv):
parser = argparse.ArgumentParser()
parser.add_argument('--inputLoc', help='Claims Location is missing.', required=True)
parser.add_argument('--invalidSubId', help='Invalid Subscriber Id Location is missing.', required=True)
parser.add_argument('--outputLoc', help='Output Location is missing.', required=True)
parser.add_argument('--runDate', help='RunDate is missing.', required=True)
return parser.parse_args(argv)

# Parse arguments
arguments = _parse_args(sys.argv[1:])
input_loc = arguments.inputLoc
invalid_loc = arguments.invalidSubId
output_loc = arguments.outputLoc
runDate = arguments.runDate

# Spark session
spark = SparkSession.builder
.appName("Splink")
.config("spark.sql.autoBroadcastJoinThreshold", "-1")
.getOrCreate()

# Read invalid subscriber IDs
InvalidIds = spark.read.csv(invalid_loc, sep="|")

# Read input data
df = spark.read.parquet(input_loc)

# Filter out invalid subscriber IDs
sdf = df.alias('c')
.join(InvalidIds, df.subscriber_id == InvalidIds._c0, "leftanti")
.select("c.*").distinct()

# Define SQL case expressions using native Spark SQL functions
name_inversion = """
CASE
WHEN firstname_l IS NULL OR firstname_r IS NULL OR lastname_l IS NULL OR lastname_r IS NULL THEN -1
WHEN firstname_l = lastname_r AND firstname_r = lastname_l THEN 2
WHEN levenshtein(firstname_l, lastname_r) < 2 AND levenshtein(firstname_r, lastname_l) < 2 THEN 1
ELSE 0 END
"""

street_exp = """
CASE
WHEN street_l IS NULL OR street_r IS NULL THEN -1
WHEN levenshtein(street_l, street_r) <= 2 THEN 5
WHEN levenshtein(street_l, street_r) <= 4 THEN 4
WHEN levenshtein(street_l, street_r) <= 6 THEN 3
WHEN levenshtein(street_l, street_r) <= 8 THEN 2
WHEN levenshtein(street_l, street_r) <= 10 THEN 1
ELSE 0 END
"""

first_name_jaro_exp = """
CASE
WHEN firstname_l IS NULL OR firstname_r IS NULL THEN -1
WHEN levenshtein(firstname_l, firstname_r) <= 2 THEN 4
WHEN levenshtein(firstname_l, firstname_r) <= 4 THEN 3
WHEN levenshtein(firstname_l, firstname_r) <= 6 THEN 2
WHEN levenshtein(firstname_l, firstname_r) <= 8 THEN 1
ELSE 0 END
"""

last_name_jaro_exp = """
CASE
WHEN lastname_l IS NULL OR lastname_r IS NULL THEN -1
WHEN levenshtein(lastname_l, lastname_r) <= 2 THEN 3
WHEN levenshtein(lastname_l, lastname_r) <= 4 THEN 2
WHEN levenshtein(lastname_l, lastname_r) <= 6 THEN 1
ELSE 0 END
"""

birth_date_exp = """
CASE
WHEN dob_l IS NULL OR dob_r IS NULL THEN -1
WHEN dob_l = dob_r AND (substr(dob_l, -4) = '0000' OR substr(dob_r, -4) = '0000') THEN 1
WHEN dob_l = dob_r AND (substr(dob_l, -4) = '0001' OR substr(dob_r, -4) = '0001') THEN 1
WHEN dob_l = dob_r THEN 3
WHEN levenshtein(dob_l, dob_r) <= 2 THEN 2
ELSE 0 END
"""

# Splink settings
settings_dedupe_subscriber_og = {
"link_type": "dedupe_only",
"max_iterations": 25,
"blocking_rules": [
"l.state = r.state AND l.gender = r.gender AND l.dob = r.dob"
],
"comparison_columns": [
{
"col_name": "gender",
"num_levels": 2
},
{
"num_levels": 4,
"case_expression": birth_date_exp,
"custom_columns_used": ["dob"],
"custom_name": "dob_custom"
},
{
"num_levels": 4,
"case_expression": first_name_jaro_exp,
"custom_columns_used": ["firstname"],
"custom_name": "first_name_jaro_exp"
},
{
"num_levels": 4,
"case_expression": last_name_jaro_exp,
"custom_columns_used": ["lastname"],
"custom_name": "last_name_jaro_exp"
},
{
"num_levels": 4,
"case_expression": name_inversion,
"custom_columns_used": ["firstname", "lastname"],
"custom_name": "name_inversion"
},
{
"num_levels": 4,
"case_expression": street_exp,
"custom_columns_used": ["street"],
"custom_name": "street_custom"
}
],
"additional_columns_to_retain": ["subscriber_id"],
"em_convergence": 0.0001
}

# Run Splink
linker_sub_og = Splink(settings_dedupe_subscriber_og, sdf, spark)
linker_subscriber_og = linker_sub_og.get_scored_comparisons()

# Round match probability
linker_subscriber_og = linker_subscriber_og.withColumn("match_probability_val", round(col("match_probability"), 3))

# Adjust match probability for name inversion
linker_subscriber_og = linker_subscriber_og.withColumn(
'match_probability_val',
when(col("gamma_name_inversion") == 2, 0.80)
.when(col("gamma_name_inversion") == 1, 0.60)
.otherwise(col("match_probability_val"))
)

# Substring check
linker_subscriber_og = linker_subscriber_og.withColumn(
'substring_check',
when(
((col("lastname_l").contains(col("lastname_r"))) | (col("lastname_r").contains(col("lastname_l")))) &
(col("lastname_l") != col("lastname_r")) &
(col("lastname_l").isNotNull() & col("lastname_r").isNotNull()) &
(col("gamma_first_name_jaro_exp") >= 1) &
(col("gamma_dob_custom") >= 1),
1
).otherwise(0)
)

# Filter and save results
linker_subscriber_og.createOrReplaceTempView("linker_subscriber_og")
linker_subscriber_og = spark.sql("""
SELECT *
FROM linker_subscriber_og
WHERE match_probability_val > 0.001
ORDER BY unique_id_l, unique_id_r
""")

linker_subscriber_og.write.format("csv")
.option("escape", "")
.option("quote", "'")
.option("header", True)
.option("delimiter", "!")
.save(output_loc)