分组求和后的RDD排序

Question

我正在尝试运行对某些 yelp 数据进行一些分析。数据结构如下：

>>> yelp_df.printSchema()
root
 |-- business_id: string (nullable = true)
 |-- cool: integer (nullable = true)
 |-- date: string (nullable = true)
 |-- funny: integer (nullable = true)
 |-- id: string (nullable = true)
 |-- stars: integer (nullable = true)
 |-- text: string (nullable = true)
 |-- type: string (nullable = true)
 |-- useful: integer (nullable = true)
 |-- user_id: string (nullable = true)
 |-- name: string (nullable = true)
 |-- full_address: string (nullable = true)
 |-- latitude: double (nullable = true)
 |-- longitude: double (nullable = true)
 |-- neighborhoods: string (nullable = true)
 |-- open: boolean (nullable = true)
 |-- review_count: integer (nullable = true)
 |-- state: string (nullable = true)

我想计算每个州的记录数，其中包含 10 条或更多评论的整体，当前开放，并找到计数第三高的州。我先做了

>>> revDF = yelp_df.filter(yelp_df.review_count > 9)
>>> openDF = revDF.filter(revDF.open == True)
>>> openDF.groupBy("state").agg({"review_count":"sum"}).collect()

这给了这个

[Row(state=u'MN', SUM(review_count#16)=3470), Row(state=u'GA', SUM(review_count#16)=5764), Row(state=u'TX', SUM(review_count#16)=1778), Row(state=u'AZ', SUM(review_count#16)=72214), Row(state=u'NY', SUM(review_count#16)=4081), Row(state=u'OR', SUM(review_count#16)=2125), Row(state=u'ID', SUM(review_count#16)=429), Row(state=u'CA', SUM(review_count#16)=1876), Row(state=u'CO', SUM(review_count#16)=6720), Row(state=u'WA', SUM(review_count#16)=525), Row(state=u'LA', SUM(review_count#16)=8394)]

现在将其存储到 summedDF 后，

summedDF.sort(summedDF.state.desc()).collect()

按状态排序就好了，但是（不出所料）

summedDF.sort(summedDF.SUM(review_count#16).desc()).collect()

不起作用。实际上，它甚至运行。我有正确数量的括号，但它没有执行，而是转到下一行 ... 之前，等待新输入。

如何进行排序，不执行是怎么回事？ #16 是怎么回事？

Answer 1

编辑： 为 pyspark 添加了版本。

我建议您将代码重构为：

val finalDF = yelp_df
  .where(col("review_count") > 9 && col("open") === true)
  .groupBy("state")
  .agg(sum("review_count").as("sum_column"))
  .sort(col("sum_column").desc)

也许我们可以适应 pyspark：

from pyspark.sql.functions import *
finalDF = yelp_df \
    .where((col("review_count") > 9) & (col("open") == True)) \
    .groupBy("state") \
    .agg(col("state"), sum(col("review_count")).alias("sum_column")) \
    .sort(col("sum_column").desc())

现在回答你的问题：

what is going on with the non-execution? and whats with the #16?

简而言之，您尝试使用 summedDF.SUM(review_count#16) 引用该列的尝试没有成功。

sort 函数使用 Column 对象（可以通过调用 col("name") 创建），或直接使用列的名称。然而，当你做聚合时，你没有为代表总和的新列选择一个名称，所以以后引用它有点困难。为了解决这个问题，我在第四行使用了 .as("sum_column") 。

分组求和后的RDD排序

RDD sort after grouping and summing

apache-spark

pyspark

pyspark-sql