SPARK SQL等效的Qualify + Row_number语句

Question

有没有人知道Apache Spark SQL获得与标准SQL qualify（）+ rnk或row_number语句相同结果的最佳方法？

例如：

• 我有一个名为statement_data的Spark Dataframe，每个记录有12个月记录，每个记录有100个唯一的帐户记录，因此共有1200条记录
• 每个月记录都有一个名为“statement_date”的字段，可用于确定最近的记录

我希望我的最终结果是一个新的Spark Dataframe，其中包含100个唯一帐户编号中的每一个的最近3条记录（由statement_date降序确定），因此总共有300条最终记录。

在标准的Teradata SQL中，我可以执行以下操作：

select * from statement_data
qualify row_number ()
over(partition by acct_id order by statement_date desc) <= 3

Apache Spark SQL没有我所知道的独立限定功能，也许我搞砸了语法或找不到合格的文档。

如果我需要分两步执行此操作，只要这两个步骤是：

• 选择查询或替代方法，为每个account_number的记录分配排名/行编号
• 一个选择查询，我选择排名<= 3的所有记录（即选择第一，第二和第三最新记录）。

编辑1 - 7/23 2:09 pm：在安装了Spark SQL 1.4.1依赖项的Spark 1.4.1中，zero323提供的初始解决方案对我不起作用。

编辑2 - 7/23 3:24 pm：事实证明，错误与我的查询使用SQL Context对象而不是Hive Context有关。我现在能够在添加以下代码后正确运行以下解决方案来创建和使用Hive上下文：

final JavaSparkContext sc2;
final HiveContext hc2;
DataFrame df;
hc2 = TestHive$.MODULE$;
sc2 = new JavaSparkContext(hc2.sparkContext()); 
....
// Initial Spark/SQL contexts to set up Dataframes  
SparkConf conf = new SparkConf().setAppName("Statement Test");
...
DataFrame stmtSummary = 
    hc2.sql("SELECT * FROM (SELECT acct_id, stmt_end_dt, stmt_curr_bal, row_number() over (partition by acct_id order by stmt_curr_bal DESC) rank_num FROM stmt_data) tmp WHERE rank_num <= 3");

Answer 1

没有qualify（检查parser source通常很有用）但你可以使用这样的子查询：

SELECT * FROM (
    SELECT *, row_number() OVER (
        PARTITION BY acct_id ORDER BY statement_date DESC
    ) rank FROM df
 ) tmp WHERE rank <= 3

另见SPARK : failure: ``union'' expected but `(' found