title: Spark SQL window function with complex condition
body: <p>This is probably easiest to explain through example.  Suppose I have a DataFrame of user logins to a website, for instance:</p>
  
  <pre><code>scala&gt; df.show(5)
  +----------------+----------+
  |       user_name|login_date|
  +----------------+----------+
  |SirChillingtonIV|2012-01-04|
  |Booooooo99900098|2012-01-04|
  |Booooooo99900098|2012-01-06|
  |  OprahWinfreyJr|2012-01-10|
  |SirChillingtonIV|2012-01-11|
  +----------------+----------+
  only showing top 5 rows
  </code></pre>
  
  <p>I would like to add to this a column indicating when they became an active user on the site.  But there is one caveat: there is a time period during which a user is considered active, and after this period, if they log in again, their <code>became_active</code> date resets.  Suppose this period is <strong>5 days</strong>.  Then the desired table derived from the above table would be something like this:</p>
  
  <pre><code>+----------------+----------+-------------+
  |       user_name|login_date|became_active|
  +----------------+----------+-------------+
  |SirChillingtonIV|2012-01-04|   2012-01-04|
  |Booooooo99900098|2012-01-04|   2012-01-04|
  |Booooooo99900098|2012-01-06|   2012-01-04|
  |  OprahWinfreyJr|2012-01-10|   2012-01-10|
  |SirChillingtonIV|2012-01-11|   2012-01-11|
  +----------------+----------+-------------+
  </code></pre>
  
  <p>So, in particular, SirChillingtonIV's <code>became_active</code> date was reset because their second login came after the active period expired, but Booooooo99900098's <code>became_active</code> date was not reset the second time he/she logged in, because it fell within the active period.  </p>
  
  <p>My initial thought was to use window functions with <code>lag</code>, and then using the <code>lag</code>ged values to fill the <code>became_active</code> column; for instance, something starting roughly like:</p>
  
  <pre class="lang-scala prettyprint-override"><code>import org.apache.spark.sql.expressions.Window
  import org.apache.spark.sql.functions._
  
  val window = Window.partitionBy("user_name").orderBy("login_date")
  val df2 = df.withColumn("tmp", lag("login_date", 1).over(window))
  </code></pre>
  
  <p>Then, the rule to fill in the <code>became_active</code> date would be, if <code>tmp</code> is <code>null</code> (i.e., if it's the first ever login) or if <code>login_date - tmp &gt;= 5</code> then <code>became_active = login_date</code>; otherwise, go to the next most recent value in <code>tmp</code> and apply the same rule.  This suggests a recursive approach, which I'm having trouble imagining a way to implement.  </p>
  
  <p>My questions: Is this a viable approach, and if so, how can I "go back" and look at earlier values of <code>tmp</code> until I find one where I stop?  I can't, to my knowledge, iterate through values of a Spark SQL <code>Column</code>.  Is there another way to achieve this result?  </p>