title: Filtering rows based on column values in spark dataframe scala
body: <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>