How to filter out all the duplicate phone numbers in spark/sql

user_10801295 0
Viewed 97

Input Table:

ID Phone Number
1 202-555-0188:202-555-0132
2 202-555-0104:202-555-0132
3 202-555-0104:202-555-0162
4 202-555-0181:202-555-0188:202-556-7191
5 518-555-0110

Given above table how can you filter out all the duplicate numbers in Phone Number Column? Numbers must be unique across the entire column, not just for that particular cell.

  1. How to use RDD to achieve the same?
  2. Can it be done using just SQL?

Expected Output:

ID Phone Number
1 202-555-0188:202-555-0132
2 202-555-0104
3 202-555-0162
4 202-555-0181:202-556-7191
5 518-555-0110

Explanation:

  1. 202-555-0132 - is a duplicate in row 2 from row 1, thus removed
  2. 202-555-0104 - is a duplicate in row 3 from row 2, thus removed
  3. 202-555-0188 - is a duplicate in row 4 from row 1, thus removed

Assumption: There will be atleast one unique number per row.

apache-spark sql
2 Answers

create an array from the string, deduplicate and recreate the string (if needed):

import spark.implicits._
import org.apache.spark.sql.functions._
val data = Seq((1,"202-555-0188:202-555-0132"),(2,"202-555-0104:202-555-0132"))
val df=sc.parallelize(data).toDF("id","phone")
val df2=df.withColumn("ps",split($"phone",":"))
val df3=df2.withColumn("p",explode ($"ps")).drop($"ps").drop($"phone")
val df4=df3.dropDuplicates("p")
val df5=df4.groupBy($"id").agg(collect_list($"p").as("ps"))
df5.withColumn("phones",array_join($"ps",":")).drop($"ps").show(10,false)

+---+-------------------------+
|id |phones                   |
+---+-------------------------+
|1  |202-555-0188:202-555-0132|
|2  |202-555-0104             |
+---+-------------------------+
user_10186590

Considering Spark Dataset API it's always better to use case class for schema definition.

  val spark = SparkSession.builder().appName("test").master("local[*]").getOrCreate()
  import spark.implicits._
  case class User(id: Int, phone_no: String)
  val userDS = Seq[User](User(1,"202-555-0188:202-555-0132"),User(2,"202-555-0104:202-555-0132"),User(3,"202-555-0104:202-555-0162"),User(4, "202-555-0181:202-555-0188:202-556-7191"),User(5, "518-555-0110")).toDS
  val parsedDS = userDS.map{user => (user.id, user.phone_no.split(":"))}
  val outDS = parsedDS.select($"_1".as("ID"), explode($"_2")
    .as("PhoneNo")).dropDuplicates("PhoneNo")
    .groupBy("ID").agg(concat_ws(":",collect_set($"PhoneNo")).as("Phone Number"))
    .orderBy("ID")
  outDS.show(false)
    
+---+-------------------------+
|ID |Phone Number             |
+---+-------------------------+
|1  |202-555-0132:202-555-0188|
|2  |202-555-0104             |
|3  |202-555-0162             |
|4  |202-555-0181:202-556-7191|
|5  |518-555-0110             |
+---+-------------------------+
user_54622590
Related