7

Question in Brief:

For a more direct query, i want to run over all the rows sequentially, and assign some values to some variables (a, b, c), based on certain conditions for the specific row, then i would assign the value of 1 of these variables into a column of that particular row.

Detailed:

I want to update a column value in the data frame in spark. The update will be conditional, where in I will run a loop on row and update a column based on the values of the other columns of that row.

I tried to use withColumn approach but got error. Please suggest any other approach. The resolution of the withColumn approach will also be of great help.

Table:

var table1 = Seq((11, 25, 2, 0), (42, 20, 10, 0)).toDF("col_1", "col_2", "col_3", "col_4")
table1.show()

Schema:

+-----+-----+-----+-----+
|col_1|col_2|col_3|col_4|
+-----+-----+-----+-----+
|   11|   25|    2|    0|
|   42|   20|   10|    0|
+-----+-----+-----+-----+

I have tried 2 approaches here:

  1. withColumn
  2. i("col_4") = adj_c

In the below code, the variables initialised at different locations need to be placed in this way only, as per the conditions

Code:

for(i <- table1.rdd.collect()) {
    if(i.getAs[Int]("col_1") > 0) {
       var adj_a = 0
       var adj_c = 0
        if(i.getAs[Int]("col_1") > (i.getAs[Int]("col_2") + i.getAs[Int]("col_3"))) {
            if(i.getAs[Int]("col_1") < i.getAs[Int]("col_2")) {
                adj_a = 10
                adj_c = 2
            }
            else {
                adj_a = 5
            }
        }
        else {
            adj_c = 1
        }
        adj_c = adj_c + i.getAs[Int]("col_2")
        table1.withColumn("col_4", adj_c)
         //i("col_4")  = adj_c
    }
}

Error in 1st case:

table1.withColumn("col_4", adj_c)

<console>:80: error: type mismatch;
 found   : Int
 required: org.apache.spark.sql.Column
               table1.withColumn("col_4", adj_c)
                                          ^

I also tried to use col(adj_c) here, but it started failing with

<console>:80: error: type mismatch;
 found   : Int
 required: String
               table1.withColumn("col_4", col(adj_c))
                                              ^

Error in 2nd case:

(i("col_4") = adj_c)

<console>:81: error: value update is not a member of org.apache.spark.sql.Row
                i("col_4")  = adj_c
                ^

I want the output table to be:

+-----+-----+-----+-----+
|col_1|col_2|col_3|col_4|
+-----+-----+-----+-----+
|   11|   25|    2|    1|
|   42|   20|   10|    5|
+-----+-----+-----+-----+

Please suggest the possible solutions and revert in case of any doubt with the question.

Please help me with this as i am stuck with issue. Any kind of suggestion will be very helpful.

5
  • Can you please add an example of input with its desired output?
    – Nir Hedvat
    Jun 18, 2019 at 11:42
  • The input and the output are given in the form of tables, with columns (col_1, col_2, col_3, col_4), where in value of col_4 is being modified Jun 18, 2019 at 11:51
  • Can you please explain what your conditions are in "The update will be conditional"? @Yashi Jun 21, 2019 at 14:05
  • @C.S.ReddyGadipally the conditions are mainly based on the values of the other columns of that row. Example: (col_1 + col_2) > col_3 Jun 22, 2019 at 19:21
  • Use foreach method on dataframe and call your method and inside that method use case class to convert the data frame row to object and perform what ever operations you want. I can share the example if you want.
    – Suresh
    Jun 27, 2019 at 14:47

3 Answers 3

5

You should use a when function instead of such complicated syntax, also there is no need for an explicit loop, Spark handles it itself. When you perform a withColumn it is applied to each row

table1.withColumn("col_4", when($"col_1" > $"col_2" + $"col_3", 5).otherwise(1)).show

QUICK TEST:

INPUT

table1.show

-----+-----+-----+-----+
|col_1|col_2|col_3|col_4|
+-----+-----+-----+-----+
|   11|   25|    2|    0|
|   42|   20|   10|    0|
+-----+-----+-----+-----+

OUTPUT

table1.withColumn("col_4", when($"col_1" > $"col_2" + $"col_3", lit(5)).otherwise(1)).show
+-----+-----+-----+-----+
|col_1|col_2|col_3|col_4|
+-----+-----+-----+-----+
|   11|   25|    2|    1|
|   42|   20|   10|    5|
+-----+-----+-----+-----+
6
  • Here I am initializing adj_c with an integer, but actually this will be computed by an expression. Also, it will include nested if conditions in this. In these 2 cases, this approach can cause issue. Jun 18, 2019 at 11:56
  • You can nest the number of when conditions you want, also you can set the new column value with an expression depending on other columns or in other variables
    – SCouto
    Jun 18, 2019 at 11:59
  • There are multiple variable assignments in different conditions, which are being used further. Can we do variable assignments in when statement? Jun 18, 2019 at 12:13
  • Yes, you can. But i think you should update your question with a more accurate example of your actual problem
    – SCouto
    Jun 18, 2019 at 12:32
  • i have tried to edit the question to bring a little more clarity. The edit is mainly in the sample data. Jun 18, 2019 at 14:31
5

UDF can be used with any custom logic for caluclate column value, like:

val calculateCol4 = (col_1:Int, col_2:Int, col_3:Int)  =>
  if (col_1 > 0) {

    var adj_a = 0
    var adj_c = 0
    if (col_1 > col_2 + col_3) {
      if (col_1 < col_2) {
        adj_a = 10
        adj_c = 2
      }
      else {
        adj_a = 5
      }
    }
    else {
      adj_c = 1
    }
    println("adj_c: "+adj_c)
    adj_c = adj_c + col_2
    // added for return correct result
    adj_c
  }
  // added for return correct result
  else 0

val col4UDF = udf(calculateCol4)
table1.withColumn("col_4",col4UDF($"col_1", $"col_2", $"col_3"))
2
  • The approach works well for me, as for 1 row edits. I just came through another situation wherein i need to change previous rows based on the some conditions. Its more like, if value of col_1 exceed 20, i will add the remaining amount in the previous row. It would be really helpful if you can something for this as well. Jun 26, 2019 at 9:48
  • Guess, previous rows change is another question, not related to current. Looks like this can be achived with Window functions, but more details is required. Also if "previous rows change" will be in another question, more people can help with this, not only me.
    – pasha701
    Jun 26, 2019 at 10:06
3
+25

using spark.sql, more easy to read and understand -

scala> var table1 = Seq((11, 25, 2, 0), (42, 20, 10, 0)).toDF("col_1", "col_2", "col_3", "col_4")
table1: org.apache.spark.sql.DataFrame = [col_1: int, col_2: int ... 2 more fields]

scala> table1.show()
+-----+-----+-----+-----+
|col_1|col_2|col_3|col_4|
+-----+-----+-----+-----+
|   11|   25|    2|    0|
|   42|   20|   10|    0|
+-----+-----+-----+-----+

scala> table1.createOrReplaceTempView("table1")


scala> val result = spark.sql(s""" select col_1,
     |                                    col_2,
     |                                    col_3,
     |                                    CASE WHEN col_1 > (col_2 + col_3)
     |                                           THEN 5
     |                                         ELSE   1
     |                                    END as col_4
     |                              from  table1 """)
result: org.apache.spark.sql.DataFrame = [col_1: int, col_2: int ... 2 more fields]


scala> result.show(false)
+-----+-----+-----+-----+
|col_1|col_2|col_3|col_4|
+-----+-----+-----+-----+
|11   |25   |2    |1    |
|42   |20   |10   |5    |
+-----+-----+-----+-----+

Hope this is helpful.

2
  • actually the original conditions are too many, so did not want to use query for this. But thanks for the answer Jun 26, 2019 at 9:45
  • 2
    You can easily add multiple conditions in case statement and it will be more readable compare to dataframe/dataset operations.
    – Ajay Ahuja
    Jun 26, 2019 at 11:51

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.