CODEWITHSUNDEEP
apache-sparkpysparkapache-spark-sql
Salam
Salam

Reputation: 33

Spark Scala: update dataframe column's value from another dataframe

a = 

+------------+------------+------+
|        Name| Nationality|Salary|
+------------+------------+------+
|    A. Abbas|        Iraq|   €2K|
| A. Abdallah|      France|   €1K|
|A. Abdennour|     Tunisia|  €31K|

b = 

+------------+------------+
|        Name|Salary      |
+------------+------------+
|    A. Abbas|€4K         |
| A. Abdallah|€1K         |
|A. Abdennour|€33K        |

the expected updatedDF should look like below: 

+------------+------------+------+
|        Name| Nationality|Salary|
+------------+------------+------+
|    A. Abbas|        Iraq|   €4K|
| A. Abdallah|      France|   €1K|
|A. Abdennour|     Tunisia|  €33K|

I tried in spark scala code like : 

updatedDF = a.join(b, Seq("Name"), "inner")
updatedDF.show()

But I have duplication in my output after doing join. how I can merge between tow data frames with out duplication ?

Upvotes: 1

Views: 1166

Answers (2)

Hanan Atallah
Hanan Atallah

Reputation: 120

If you have duplication, that means name column is not unique. I suggest to try append index column to be used in join, then drop it:

    // Add index now...
    a = addColumnIndex(a).withColumn("index", monotonically_increasing_id)
    println("1- a count: " + a.count())

    // Add index now...
    b = addColumnIndex(b).withColumn("index", monotonically_increasing_id)
    println("b count: " + b.count())

    def addColumnIndex(df: DataFrame) = {
        spark.sqlContext.createDataFrame(
            df.rdd.zipWithIndex.map {
                case (row, index) => Row.fromSeq(row.toSeq :+ index)
            },
            StructType(df.schema.fields :+ StructField("index", LongType, false)))
    }

    ab = a.join(b, Seq("index", "Name"), "inner").drop(a.col("Salary")).drop(a.col("index"))

    println("3- ab count: " + ab.count())

Upvotes: 0

Chandan Ray
Chandan Ray

Reputation: 2091

val a = sc.parallelize(List(("A. Abbas","Iraq","2K"),("A. Abdallah","France","1K"),("A. Abdennour","Tunisia","31K"))).toDF("Name","Nationality","Salary")
val b = sc.parallelize(List(("A. Abbas","4K"),("A. Abdallah","1K"),("A. Abdennour","33K"))).toDF("Name","Salary")
b.join(a,Seq("Name"),"inner").drop(a.col("Salary")).show

enter image description here

Upvotes: 2

Related Questions