apache-spark - SQL查询转Spark Dataframe结构化数据处理

Question

我想在spark dataframe中转换下面的query:

sqlContext.sql("SELECT d.dep_name,count(*) FROM employees e,department d WHERE e.dep_id = d.dep_id GROUP BY d.dep_name HAVING count(*) >= 2").show  

输出:

+---------+---+                                                                 
| dep_name|_c1|
+---------+---+
|  FINANCE|  3|
|    AUDIT|  5|
|MARKETING|  6|

我尝试使用以下查询:

scala> finalEmployeesDf.as("df1").join(depDf.as("df2"), $"df1.dep_id" === $"df2.dep_id").select($"dep_name").groupBy($"dep_name").count.show()
+---------+-----+                                                               
| dep_name|count|
+---------+-----+
|  FINANCE|    3|
|    AUDIT|    5|
|MARKETING|    6|
+---------+-----+  

我知道这 不正确 因为假设我们有一个情况，我们只有一个部门条目，那么它也会列在这些结果中，但我希望结果仅在以下情况下显示计数大于 2。那么我该如何实现呢？？？我尝试使用谷歌搜索，但在这种情况下没有任何帮助。

Answer 1

您的分组和聚合部分有误。您需要选择所有相关列，分组依据并按相关一次聚合。这是代表正确方法的未经测试的代码:

finalEmployeesDf.as("df1")
 .join(depDf.as("df2"), $"df1.dep_id" === $"df2.dep_id")
 .select($"dep_name")
 .groupBy($"dep_name")
 .agg(count($"dep_name").as("cnt"))
 .filter($"cnt" > 2)
 .show()

一般的建议是尝试将 API 调用分成多行，这样可以大大更容易阅读和理解。