计算 Spark 数据框中的单词数

在不使用 SQL 的 REPLACE() 函数的情况下，如何找到 Spark 数据帧的列中的单词数？下面是我正在使用的代码和输入，但 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|

使用 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 column.

每行计数

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

• Use pyspark.sql.functions.split() http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.html#pyspark.sql.functions.split将字符串分解为列表
• Use pyspark.sql.functions.size() http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.html#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() http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.html#pyspark.sql.functions.sum:

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

统计每个单词出现的次数

如果您想要整个 DataFrame 中每个单词的计数，您可以使用split() and pyspark.sql.function.explode() http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.html#pyspark.sql.functions.explode随后是一个groupBy and 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|
#+---------+-----+