sparksql分组后topN（JAVA）

<!-- spark -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.10</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.10</artifactId>
<version>1.6.0</version>
</dependency>

public class RowNumberTopN {
public static void main(String[] args) {
SparkConf conf = new SparkConf().setAppName("RowNumberTopN").setMaster("local");

JavaSparkContext sc = new JavaSparkContext(conf);
HiveContext hiveContext = new HiveContext(sc.sc());
hiveContext.sql("drop table if exists sales");
//创建销售表
hiveContext.sql("create table if not exists sales(product STRING,category STRING,revenue BIGINT)");
hiveContext.sql("LOAD DATA LOCAL INPATH '/user/local/spark/resources/sales.txt' INTO TABLE sales");
4000
//进行row_number分组
DataFrame topnDF = hiveContext.sql("SELECT product,category,revenue FROM (\n" +
"SELECT product,category,revenue,row_number() OVER (PARTITION BY category ORDER BY revenue DESC) rank\n" +
"FROM sales\n" +
") topn_sales where rank<=3");
hiveContext.sql("drop table if exists top3_sales");
topnDF.saveAsTable("top3_sales");
sc.close();
}
}