Check particular identifier is present in the other data frame or not

user_4349844 0
Viewed 71
val df1 = Seq(("[1,10,20]", "bat","43243"),("[20,4,10]","mouse","4324432"),("[30,20,3]", "horse","4324234")).toDF("id", "word","userid") 

val df2 = Seq((1, "raj", "name"),(2, "kiran","name"),(3,"karnataka","state"),(4, "Andrapradesh","state")).toDF("id", "name", "code")

Explanation:

I have two dataframes df1 & df2. df1 is having id column which is having a list of ids.

I need to check any one of the ids is present in the df2 dataframe.

Conditions

if id is present in the df2 id column and if the code is state then get the name of the particular id from the df2 and create a new data frame with name column

Expected Output

id        |     word    |  userid   | name

-------------------------+-----------------------

[30,20,3] |  "horse"    | "4324234" | "karnataka"  
[20,4,10] |  "mouse"    | "4324432" | "Andrapradesh"
apache-spark-sql apache-spark dataframe scala
2 Answers

You can flatten the id column first by converting it to an array and applying explode. Then you can apply a normal join operation between the DataFrames.

For example:

val df1 = Seq(("[1,10,20]", "bat","43243"),("[20,4,10]","mouse","4324432"),("[30,20,3]", "horse","4324234")).toDF("id", "word","userid") 
val df2 = Seq((1, "raj", "name"),(2, "kiran","name"),(3,"karnataka","state"),(4, "Andrapradesh","state")).toDF("id", "name", "code")

val flattenDf1 = df1.
  select(
    col("id"),
    expr("""split(regexp_replace(id, "\\[|\\]",""), ",")""").as("idArray"), col("word"),
    col("userid")).
  withColumn("id_", explode(col("idArray"))).
  drop("idArray")

df2.as("df2").
  join(
    flattenDf1.as("df1"),
    col("df2.id") === col("df1.id_")).
  filter("code = 'state'").
  select("df1.id", "word", "userid", "name").
  show
// Result: 
// +---------+-----+-------+------------+
// |       id| word| userid|        name|
// +---------+-----+-------+------------+
// |[30,20,3]|horse|4324234|   karnataka|
// |[20,4,10]|mouse|4324432|Andrapradesh|
// +---------+-----+-------+------------+

I hope it helps.

user_25167830

You could just use UDF as a condition in the join:

val arrayJoin = udf { 
   (a: WrappedArray[Int], v: Int) => a.contains(v) 
}

val result = df1
      .join(df2.as("df2"), arrayJoin(df2("id"), df1("id"))) //join using udf
      .drop("df2.id", "df2.code") //drop unnecessary columns
user_13033240
Related