如何在Spark SQL中的爆炸字段上做GROUP BY？

Question

• 齐柏林0.6
• 星火1.6
• SQL

我试图找到一些微博的前20名出现的词。 filtered包含每个推文的单词数组。以下：

select explode(filtered) AS words from tweettable 

列出了每个单词你所期望的，但我要的是让每个单词的数量在所有的鸣叫，然后显示这些排名前20位。下面的作品，但我需要为此在SQL：

df.select(explode($"filtered").as("value")) 
    .groupBy("value") 
    .count() 
    .sort(desc("count")) 
    .show(20, false) 

我试图GROUP BY上words，filtered，并且explode(filtered)但所有给了错误。

Answer 1

可以在FROM语句中使用subqueries：

SELECT value, count(*) AS count 
FROM (SELECT explode(filtered) AS value 
     FROM tweettable) AS temp 
GROUP BY value 
ORDER BY count DESC 

Answer 2

下面的代码会给你全面的了解，以达到你所期待的东西。测试火花（1.6）

val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) 
import hiveContext.implicits._ 

val lst = List(Seq("Hello","Hyd","Hello","Mumbai"),Seq("Hello","Mumbai"),Seq("Hello","Delhi","Hello","Banglore")) 
case class Tweets(filtered: Seq[String]) 
val df = sc.parallelize(lst).map(x=>Tweets(x)).toDF 

import org.apache.spark.sql.functions.{explode} 
import org.apache.spark.sql.functions.count 
df.select(explode($"filtered").as("value")).groupBy("value").agg(count("*").alias("cnt")).orderBy('cnt.desc).show(20,false) 

或者，您可以使用窗口功能。

val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) 
import hiveContext.implicits._ 

val lst = List(Seq("Hello","Hyd","Hello","Mumbai"),Seq("Hello","Mumbai"),Seq("Hello","Delhi","Hello","Banglore")) 
case class Tweets(filtered: Seq[String]) 
val df = sc.parallelize(lst).map(x=>Tweets(x)).toDF 

import org.apache.spark.sql.expressions.Window 
import org.apache.spark.sql.functions._ 
val w = org.apache.spark.sql.expressions.Window.orderBy('cnt.desc) 

df.select(explode($"filtered").as("value")).groupBy("value").agg(count("*").alias("cnt")).withColumn("filteredrank", rank.over(w)).filter(col("filteredrank") <= 20).show()