1

我正在尝试对一些 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 怎么了？

4

1 回答 1

3

编辑：为 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())

现在回答你的问题：

不执行是怎么回事？#16 怎么了？

简而言之，您尝试使用 using 来引用该列是summedDF.SUM(review_count#16)行不通的。

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

于 2016-05-04T17:48:34.360 回答