Create new Dataframe column based on existing Dataframes in Spark

Sanjay Source

There are two DF's , I need to populate a new column in DF1 say Flag on below conditions.

DF1
    +------+-------------------+
    ||AMOUNT|Brand             | 
    +------+-------------------+
    | 47.88|          Parle    |
    | 40.92|          Parle    |
    | 83.82|          Parle    |
    |106.58|          Parle    |
    | 90.51|          Flipkart |
    | 11.48|          Flipkart |
    | 18.47|          Flipkart |
    | 40.92|          Flipkart |
    |  30.0|          Flipkart |
    +------+-------------------+


DF2

+--------------------+-------+----------+
|       Brand        |   P1  |   P2     |
+--------------------+-------+----------+
|               Parle| 37.00 |  100.15  |
|            Flipkart|  10.0 |  30.0    |
+--------------------+-------+----------+

if the amount of say Brand Parle in DF1 is less than P1 value (Amount < P1) in DF2 for the brand "Parle", then flag will be low, if P1 >= amount <= P2 than flag will be mid and if Amount > P2 then high, Similarly for other merchants too.

the DF1 is having very huge data and DF2 is very small.

expected output

+------+-------------------+----------------+
||AMOUNT|Brand             |        Flag    |
+------+-------------------+----------------+
| 47.88|          Parle    |   mid          |
| 40.92|          Parle    |   mid          |
| 83.82|          Parle    |   mid          |
|106.58|          Parle    |   high         |
| 90.51|          Flipkart |   high         |
| 11.48|          Flipkart |   mid          |
| 18.47|          Flipkart |   mid          |
| 40.92|          Flipkart |   high         |
|  30.0|          Flipkart |   mid          |
+------+-------------------+----------------

I know that I can do a join and get the results, but how should i frame to logic in spark.

scalaapache-spark

Answers

answered 5 months ago Ramesh Maharjan #1

Simple left join and nested when inbuilt functions should get your desired result as

import org.apache.spark.sql.functions._
df1.join(df2, Seq("Brand"), "left")
  .withColumn("Flag", when(col("AMOUNT") < col("P1"), "low").otherwise(
    when(col("AMOUNT") >= col("P1") && col("AMOUNT") <= col("P2"), "mid").otherwise(
      when(col("AMOUNT") > col("P2"), "high").otherwise("unknown"))))
    .select("AMOUNT", "Brand", "Flag")
  .show(false)

which should give you

+------+--------+----+
|AMOUNT|Brand   |Flag|
+------+--------+----+
|47.88 |Parle   |mid |
|40.92 |Parle   |mid |
|83.82 |Parle   |mid |
|106.58|Parle   |high|
|90.51 |Flipkart|high|
|11.48 |Flipkart|mid |
|18.47 |Flipkart|mid |
|40.92 |Flipkart|high|
|30.0  |Flipkart|mid |
+------+--------+----+

I hope the answer is helpful

answered 5 months ago Hayk Hakobyan #2

I think with udf is also doable.

val df3 = df1.join(df2, Seq("Brand"), "left")
import org.apache.spark.sql.functions._
val mapper = udf((amount: Double, p1: Double, p2: Double) => if (amount < p1) "low" else if (amount > p2) "high" else "mid")
df3.withColumn("Flag", mapper(df3("AMOUNT"), df3("P1"), df3("P2")))
    .select("AMOUNT", "Brand", "Flag")
    .show(false)

comments powered by Disqus