Spark Scalaを使用した最後の1時間トランザクションの合計

Question

Spark-Scalaを使用してトランザクションごとに過去1時間の合計を計算します。私は三つのフィールドでデータフレームを、次のと下記のように、第4のフィールドを計算したいいる：

Customer  TimeStamp  Tr Last_1Hr_RunningSum 
Cust-1 6/1/2015 6:51:55  1  1 
Cust-1 6/1/2015 6:58:34  3  4 
Cust-1 6/1/2015 7:20:46  3  7 
Cust-1 6/1/2015 7:40:45  4  11 
Cust-1 6/1/2015 7:55:34  5  15 
Cust-1 6/1/2015 8:20:34  0  12 
Cust-1 6/1/2015 8:34:34  3  12 
Cust-1 6/1/2015 9:35:34  7  7 
Cust-1 6/1/2015 9:45:34  3  10 
Cust-2 6/1/2015 16:26:34  2  2 
Cust-2 6/1/2015 16:35:34  1  3 
Cust-2 6/1/2015 17:39:34  3  3 
Cust-2 6/1/2015 17:43:34  5  8 
Cust-3 6/1/2015 17:17:34  6  6 
Cust-3 6/1/2015 17:21:34  4  10 
Cust-3 6/1/2015 17:45:34  2  12 
Cust-3 6/1/2015 17:56:34  3  15 
Cust-3 6/1/2015 18:21:34  4  13 
Cust-3 6/1/2015 19:24:34  1  1 

私は顧客IDによって各トランザクションから一時間振り返った新しいフィールドとして「Last_1Hr_RunningSum」を計算してのいくつかを取りたいです"Tr"（Transaction filed）。例えば

1. ：得意先-1 2015年6月1日8時20分34秒では2015年6月1日7時20分46秒まで振り返ると（0 + 5 + 4 + 3）の合計を取ります= 12.
2. 各行と同じ方法1時間振り返って、その1時間にすべてのトランザクションを合計します。

入れ子になったクエリでsqlContext.sqlを実行しようとしましたが、エラーが発生しました。また、Spark-Scala SQLContextでは、ウィンドウ関数とパーティション上の行番号はサポートされていません。

Spark-Scalaで「TimeStamp」列を使用して「Tr」から最後の1時間の合計を取得するにはどうすればよいですか。

ありがとうございます。

Answer 1

私はネストされたクエリでsqlContext.sqlを実行しようとしたが、そのは

私にエラーを与え、あなたは参加してみてくださいましたか？

期待される結果を出力

df.registerTempTable("input") 

val result = sqlContext.sql(""" 
     SELECT 
      FIRST(a.Customer) AS Customer, 
      FIRST(a.Timestamp) AS Timestamp, 
      FIRST(a.Tr) AS Tr, 
      SUM(b.Tr) AS Last_1Hr_RunningSum 
     FROM input a 
     JOIN input b ON 
      a.Customer = b.Customer 
      AND b.Timestamp BETWEEN (a.Timestamp - 3600000) AND a.Timestamp 
     GROUP BY a.Customer, a.Timestamp 
     ORDER BY a.Customer, a.Timestamp 
     """) 

result.show() 

：

+--------+-------------+---+-------------------+ 
|Customer| Timestamp| Tr|Last_1Hr_RunningSum| 
+--------+-------------+---+-------------------+ 
| Cust-1|1420519915000| 1|    1.0| 
| Cust-1|1420520314000| 3|    4.0| 
| Cust-1|1420521646000| 3|    7.0| 
| Cust-1|1420522845000| 4|    11.0| 
| Cust-1|1420523734000| 5|    15.0| 
| Cust-1|1420525234000| 0|    12.0| 
| Cust-1|1420526074000| 3|    12.0| 
| Cust-1|1420529734000| 7|    7.0| 
| Cust-1|1420530334000| 3|    10.0| 
| Cust-2|1420554394000| 2|    2.0| 
| Cust-2|1420554934000| 1|    3.0| 
| Cust-2|1420558774000| 3|    3.0| 
| Cust-2|1420559014000| 5|    8.0| 
| Cust-3|1420557454000| 6|    6.0| 
| Cust-3|1420557694000| 4|    10.0| 
| Cust-3|1420559134000| 2|    12.0| 
| Cust-3|1420559794000| 3|    15.0| 
| Cust-3|1420561294000| 4|    13.0| 
| Cust-3|1420565074000| 1|    1.0| 
+--------+-------------+---+-------------------+ 

クエリを示し、サンプルデータのインデントとフォーマットを修正する必要があり