title: SparkSQL - Lag function?
body: <p>I see in this <a href="https://databricks.com/blog/2015/07/15/introducing-window-functions-in-spark-sql.html" rel="nofollow">DataBricks post</a>, there is support for window functions in SparkSql, in particular I'm trying to use the lag() window function.</p>
  
  <p>I have rows of credit card transactions, and I've sorted them, now I want to iterate over the rows, and for each row display the amount of the transaction, and the difference of the current row's amount and the preceding row's amount.</p>
  
  <p>Following the DataBricks post, I've come up with this query, but it's throwing an exception at me and I can't quite undestand why..</p>
  
  <p>This is in PySpark.. tx is my dataframe already created at registered as a temp table.</p>
  
  <pre><code>test =sqlContext.sql("SELECT tx.cc_num,tx.trans_date,tx.trans_time,tx.amt, (lag(tx.amt) OVER (PARTITION BY tx.cc_num ORDER BY  tx.trans_date,tx.trans_time ROW BETWEEN PRECEDING AND CURRENT ROW)) as prev_amt from tx")
  </code></pre>
  
  <p>and the exception (truncated)..</p>
  
  <pre><code>py4j.protocol.Py4JJavaError: An error occurred while calling o76.sql.
  : java.lang.RuntimeException: [1.67] failure: ``)'' expected but identifier OVER found
  </code></pre>
  
  <p>I'd really apprecaite any insight, this functionality is rather new and there's not a lot to go on as far as existing examples or other related posts.</p>
  
  <p><strong>Edit</strong></p>
  
  <p>I've also attempted to do this without SQL statement as below, but continue to get an error. I've used this with Hive and SQLContext, and receive the same error.</p>
  
  <pre><code>windowSpec = \
  Window \
      .partitionBy(h_tx_df_ordered['cc_num']) \
      .orderBy(h_tx_df_ordered['cc_num'],h_tx_df_ordered['trans_date'],h_tx_df_ordered['trans_time'])
  
  windowSpec.rowsBetween(-1, 0)
  
  lag_amt = \
     (lag(h_tx_df_ordered['amt']).over(windowSpec) - h_tx_df_ordered['amt'])
      tx_df_ordered.select(
      h_tx_df_ordered['cc_num'],
      h_tx_df_ordered['trans_date'],
      h_tx_df_ordered['trans_time'],
      h_tx_df_ordered['amt'],
      lag_amt.alias("prev_amt")).show()
  </code></pre>
  
  <hr>
  
  <pre><code>Traceback (most recent call last):
    File "rdd_raw_data.py", line 116, in &lt;module&gt;
      lag_amt.alias("prev_amt")).show()
    File "/opt/spark/python/pyspark/sql/dataframe.py", line 721, in select
      jdf = self._jdf.select(self._jcols(*cols))
    File "/home/brandon/anaconda/lib/python2.7/site-packages/py4j/java_gateway.py", line 813, in __call__
      answer, self.gateway_client, self.target_id, self.name)
    File "/home/brandon/anaconda/lib/python2.7/site-packages/py4j/protocol.py", line 308, in get_return_value
      format(target_id, ".", name), value)
  py4j.protocol.Py4JJavaError: An error occurred while calling o152.select.
  : org.apache.spark.sql.AnalysisException: Could not resolve window function 'lag'. Note that, using window functions currently requires a HiveContext;
      at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.failAnalysis(CheckAnalysis.scala:38)
  </code></pre>