在Spark数据框中计算单词数量

Question

如何在不使用SQL的REPLACE()函数的情况下查找Spark DataFrame列中单词的数量？以下是我正在使用的代码和输入，但replace()函数不起作用。

from pyspark.sql import SparkSession
my_spark = SparkSession \
    .builder \
    .appName("Python Spark SQL example") \
    .enableHiveSupport() \
    .getOrCreate()

parqFileName = 'gs://caserta-pyspark-eval/train.pqt'
tuesdayDF = my_spark.read.parquet(parqFileName)

tuesdayDF.createOrReplaceTempView("parquetFile")
tuesdaycrimes = spark.sql("SELECT LENGTH(Address) - LENGTH(REPLACE(Address, ' ', ''))+1 FROM parquetFile")

print(tuesdaycrimes.show())


+-------------------+--------------+--------------------+---------+----------+--------------+--------------------+-----------+---------+
|              Dates|      Category|            Descript|DayOfWeek|PdDistrict|    Resolution|             Address|          X|        Y|
+-------------------+--------------+--------------------+---------+----------+--------------+--------------------+-----------+---------+
|2015-05-14 03:53:00|      WARRANTS|      WARRANT ARREST|Wednesday|  NORTHERN|ARREST, BOOKED|  OAK ST / LAGUNA ST| -122.42589|37.774597|
|2015-05-14 03:53:00|OTHER OFFENSES|TRAFFIC VIOLATION...|Wednesday|  NORTHERN|ARREST, BOOKED|  OAK ST / LAGUNA ST| -122.42589|37.774597|
|2015-05-14 03:33:00|OTHER OFFENSES|TRAFFIC VIOLATION...|Wednesday|  NORTHERN|ARREST, BOOKED|VANNESS AV / GREE...| -122.42436|37.800415|

- Hrishikesh Sarma

---

df.select("columnName").count() 应该足够。 - Ramesh Maharjan

@gordonlinoff：它会提示出一个错误，说这是一个未定义的函数。 - Hrishikesh Sarma

你能提供一个可重现的例子吗？ - mtoto

你需要编写一个udf函数来实现。正如@mtoto所建议的，您需要更新问题，包括输入、已尝试的方法和期望输出。你可以这样做吗？如果你这样做了，我相信你会立刻得到很多答案。 - Ramesh Maharjan

@RameshMaharjan 刚刚添加了更多信息。感谢您的帮助。 - Hrishikesh Sarma

Answer 1

使用pyspark DataFrame函数计算单词数量有多种方法，具体取决于您所需求的内容。

创建示例数据

import pyspark.sql.functions as f
data = [
    ("2015-05-14 03:53:00", "WARRANT ARREST"),
    ("2015-05-14 03:53:00", "TRAFFIC VIOLATION"),
    ("2015-05-14 03:33:00", "TRAFFIC VIOLATION")
]

df = sqlCtx.createDataFrame(data, ["Dates", "Description"])
df.show()

在这个例子中，我们将计算Description列中的单词数。

每行计数

如果您想要获取指定列每行中单词的数量，您可以使用withColumn()创建一个新列，并执行以下操作：

• 使用pyspark.sql.functions.split()将字符串拆分成列表
• 使用pyspark.sql.functions.size()计算列表的长度

例如：

df = df.withColumn('wordCount', f.size(f.split(f.col('Description'), ' ')))
df.show()
#+-------------------+-----------------+---------+
#|              Dates|      Description|wordCount|
#+-------------------+-----------------+---------+
#|2015-05-14 03:53:00|   WARRANT ARREST|        2|
#|2015-05-14 03:53:00|TRAFFIC VIOLATION|        2|
#|2015-05-14 03:33:00|TRAFFIC VIOLATION|        2|
#+-------------------+-----------------+---------+

统计所有行的单词总数

如果您想要计算整个DataFrame中该列的单词总数，可以使用pyspark.sql.functions.sum()：

df.select(f.sum('wordCount')).collect() 
#[Row(sum(wordCount)=6)]

统计每个单词出现的次数

如果您想获得整个DataFrame中每个单词的计数，可以使用split()和pyspark.sql.function.explode()，然后跟随groupBy和count()。

df.withColumn('word', f.explode(f.split(f.col('Description'), ' ')))\
    .groupBy('word')\
    .count()\
    .sort('count', ascending=False)\
    .show()
#+---------+-----+
#|     word|count|
#+---------+-----+
#|  TRAFFIC|    2|
#|VIOLATION|    2|
#|  WARRANT|    1|
#|   ARREST|    1|
#+---------+-----+

- pault

---

我该如何使用统计每个单词出现次数的代码，但同时包含停用词？ - Hefe

Answer 2

您可以使用pyspark的API函数split和size来完成此操作（以下是示例）：

sqlContext.createDataFrame([['this is a sample address'],['another address']])\
.select(F.size(F.split(F.col("_1"), " "))).show()

Below is Output:-
+------------------+
|size(split(_1,  ))|
+------------------+
|                 5|
|                 2|
+------------------+

- Rakesh Kumar

---

这里的F是指数据框吗？ - Hrishikesh Sarma

F 就像 from pyspark.sql import functions as F 中的 F。 - eliasah

Answer 3

你可以定义一个 udf 函数，如下所示：

def splitAndCountUdf(x):
    return len(x.split(" "))

from pyspark.sql import functions as F
countWords = F.udf(splitAndCountUdf, 'int')

并使用.withColumn函数进行调用。

tuesdayDF.withColumn("wordCount", countWords(tuesdayDF.address))

如果您想要单词的不同计数，您可以更改udf函数以包括set。

def splitAndCountUdf(x):
    return len(set(x.split(" ")))

from pyspark.sql import functions as F
countWords = F.udf(splitAndCountUdf, 'int')

- Ramesh Maharjan

Answer 4

使用Spark SQL

SELECT word, count(*)
FROM
    (SELECT explode(split(Description, ' ')) AS word FROM mytable)
GROUP BY 1
ORDER BY 2 DESC

完整示例

data = [
    ("2015-05-14 03:53:00", "WARRANT ARREST"),
    ("2015-05-14 03:53:00", "TRAFFIC VIOLATION"),
    ("2015-05-14 03:33:00", "TRAFFIC VIOLATION")
]

df = spark.createDataFrame(data, ["Dates", "Description"])
df.createOrReplaceTempView("mytable")

spark.sql("""
    SELECT word, count(*)
    FROM
        (SELECT explode(split(Description, ' ')) AS word FROM mytable)
    GROUP BY 1
    ORDER BY 2 DESC
""").show()

- phi