计算火花数据框中的字数

Posted 2023-04-17

【中文标题】计算火花数据框中的字数

【英文标题】：Count number of words in a spark dataframe

【发布时间】：2018-02-22 12:20:05

【问题描述】：

我们如何在不使用 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|

【问题讨论】：

df.select("columnName").count() 应该足够了。

replace() 方法有什么问题？

@RameshMaharjan ：这将给出总行数，而不是每行地址列中的字数

@GordonLinoff ：它给出了一个错误，它是一个未定义的函数

你能举一个可重现的例子吗？

【参考方案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|
#+---------+-----+

【讨论】：

如何使用每个单词代码的出现次数但合并停用词？

【参考方案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|
+------------------+

【讨论】：

这里的 F 指的是数据框吗？

F 与from pyspark.sql import functions as F中的一样

【参考方案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')