Scala spark to filter out reoccurring zero values

user_9906885 0
Viewed 277

I created a dataframe in spark with the following schema:

root
 |-- user_id: string (nullable = true)
 |-- rate: decimal(32,16) (nullable =true)
 |-- date: timestamp (nullable =true)
 |-- type: string (nullable = true)

Data is like this in my schema

+----------+----------+-------------+---------+
|   user_id|  rate    |date         | type    |
+----------+----------+-------------+---------+
| XO_121   | 10       |2020-04-20   | A       |         
| XO_121   | 10       |2020-04-21   |  A      |         
| XO_121   | 30       |2020-04-22   |      A  |        
| XO_121   |0         |2020-04-23   |       A |      
| XO_121   |0         |2020-04-24   |       A |        
| XO_121   |0         |2020-04-25   |      A  |        
| XO_121   |0         |2020-04-26   |       A |       
| XO_121   |5         |2020-04-27   |   A     |
| XO_121   |0         |2020-04-28   |      A  |        
| XO_121   |0         |2020-04-29   |       A |       
| XO_121   |1         |2020-04-30   |   A     |

I want to save space so I want to skip rate which has zero value but just want it's initial occurrence only other rate duplicates are allowed like you see case of 10 and they need to preserve Date order . So after applying filter my data should look like this

+----------+----------+-------------+---------+
|   user_id|  rate    |date         | type    |
+----------+----------+-------------+---------+
| XO_121   | 10       |2020-04-20   | A       |         
| XO_121   | 10       |2020-04-21   |  A      |         
| XO_121   | 30       |2020-04-22   |      A  |        
| XO_121   |0         |2020-04-23   |       A |          
| XO_121   |5         |2020-04-27   |   A     |   
| XO_121   |0         |2020-04-28   |      A  | 
| XO_121   |1         |2020-04-30   |      A  |

I'm new to spark so just want to find out way to filter . I used Rank concept but that don't work .If any body can provide solution to this problem

apache-spark-sql apache-spark scala
2 Answers

Data Preparation :

val df = Seq( ("XO_121","10","2020-04-20"),("XO_121","10","2020-04-21"),("XO_121","30","2020-04-22"),("XO_121","0","2020-04-23"),("XO_121","0","2020-04-24"),("XO_121","0","2020-04-25"),("XO_121","0","2020-04-26"),("XO_121","5","2020-04-27"),("XO_121","0","2020-04-28"),("XO_121","0","2020-04-29"),("XO_121","1","2020-04-30"))
.toDF("user_id","rate","date")

Get the previous value of rate, and check for each record "rate" === "0" && "previous_rate" === "0"

import org.apache.spark.sql.expressions.Window
val winSpec = Window.partitionBy("user_id").orderBy("date")

val finalDf = df.withColumn("previous_rate", lag("rate", 1).over(winSpec))
.filter( !($"rate" === "0" && $"previous_rate" === "0"))
.drop("previous_rate")

Output :

scala> finalDf.show
+-------+----+----------+
|user_id|rate|      date|
+-------+----+----------+
| XO_121|  10|2020-04-20|
| XO_121|  10|2020-04-21|
| XO_121|  30|2020-04-22|
| XO_121|   0|2020-04-23|
| XO_121|   5|2020-04-27|
| XO_121|   0|2020-04-28|
| XO_121|   1|2020-04-30|
+-------+----+----------+

Now you can apply orderBy($"date") or orderBy($"userd_id",$"date") which ever is applicable for you.

user_70945200

You can use row_number() instead of Rank as below

_w = W.partitionBy("col2").orderBy("col1")
df = df.withColumn("rnk", F.row_number().over(_w))
df = df.filter(F.col("rnk") == F.lit("1"))
df.show()
+------+----+---+
|  col1|col2|rnk|
+------+----+---+
|XO_121|   0|  1|
|XO_121|  10|  1|
|XO_121|  30|  1|
|XO_121|  20|  1|
|XO_121|  40|  1|
+------+----+---+

Also , you can use first() in case you know there is only repetition on value 0

df = df.groupBy("col1","col2").agg(F.first("col2").alias("col2")).orderBy("col2")
df.show()
+------+----+----+
|  col1|col2|col2|
+------+----+----+
|XO_121|   0|   0|
|XO_121|  10|  10|
|XO_121|  20|  20|
|XO_121|  30|  30|
|XO_121|  50|  50|
+------+----+----+
user_51082200
Related