Filtering rows based on column values in spark dataframe scala
<p>I have a dataframe(spark):</p>

<pre><code>id  value 
3     0
3     1
3     0
4     1
4     0
4     0
</code></pre>

<p>I want to create a new dataframe:</p>

<pre><code>3 0
3 1
4 1
</code></pre>

<p>Need to remove all the rows after 1(value) for each id.I tried with window functions in spark dateframe(Scala). But couldn't able to find a solution.Seems to be I am going in a wrong direction.</p>

<p>I am looking for a solution in Scala.Thanks</p>

<p>Output using monotonically_increasing_id</p>

<pre><code> scala&gt; val data = Seq((3,0),(3,1),(3,0),(4,1),(4,0),(4,0)).toDF("id", "value")
data: org.apache.spark.sql.DataFrame = [id: int, value: int]

scala&gt; val minIdx = dataWithIndex.filter($"value" === 1).groupBy($"id").agg(min($"idx")).toDF("r_id", "min_idx")
minIdx: org.apache.spark.sql.DataFrame = [r_id: int, min_idx: bigint]

scala&gt; dataWithIndex.join(minIdx,($"r_id" === $"id") &amp;&amp; ($"idx" &lt;= $"min_idx")).select($"id", $"value").show
+---+-----+
| id|value|
+---+-----+
|  3|    0|
|  3|    1|
|  4|    1|
+---+-----+
</code></pre>

<p>The solution wont work if we did a sorted transformation in the original dataframe. That time the <strong>monotonically_increasing_id()</strong> is generated based on original DF rather that sorted DF.I have missed that requirement before.</p>

<p>All suggestions are welcome. </p>