Spark DataFrame统计每条评论中各句子单词数的实现问题

Hey there! Let's figure out why your sentence-level word count isn't working and fix it together. I’ve dealt with this exact scenario plenty of times, so let’s break it down step by step.

First, let's cover the common pitfalls that usually trip people up here:

• Incorrect sentence splitting (like splitting on just . which breaks words such as Dr.Jones)
• Ignoring edge cases like empty sentences, trailing spaces, or punctuation stuck to words
• Overusing custom UDFs when Spark's built-in functions are faster and more reliable

Step 1: Split Comments into Sentences Properly #

The first critical step is splitting each comment into sentences without breaking abbreviations. Using a regex with positive lookbehind ensures we split only after sentence-ending punctuation (.!?) followed by a space—this keeps terms like U.S.A. intact.

Step 2: Calculate Word Count per Sentence #

Next, we'll count words in each sentence. We’ll use Spark's built-in transform function (available in Spark 3.0+) to process each sentence in the array directly, no custom UDF needed. We’ll also trim whitespace and handle empty strings to avoid wrong counts.

Python Implementation #

Assuming your DataFrame has a column named comment storing each review:

from pyspark.sql import functions as F

# Split comments into an array of valid sentences
df_with_sentences = df.withColumn(
    "sentences",
    F.split(F.col("comment"), "(?<=[.!?])\\s+")  # Regex splits after .!? followed by space
)

# Calculate word count for each sentence in the array
df_with_word_counts = df_with_sentences.withColumn(
    "sentence_word_counts",
    F.transform(
        "sentences",
        lambda sentence: F.size(
            # Split on non-word characters, trim first to avoid empty strings
            F.split(F.trim(sentence), "\\W+")
        )
    )
)

# Optional: Explode to view each sentence and its count as separate rows
df_exploded = df_with_word_counts.select(
    "comment",
    F.explode(F.arrays_zip("sentences", "sentence_word_counts")).alias("sentence_data")
).select(
    "comment",
    F.col("sentence_data.sentences").alias("sentence"),
    F.col("sentence_data.sentence_word_counts").alias("word_count")
)

# View the final result
df_exploded.show(truncate=False)

Scala Implementation #

If you're working with Scala:

import org.apache.spark.sql.functions._

val dfWithSentences = df.withColumn(
    "sentences",
    split(col("comment"), "(?<=[.!?])\\s+")
)

val dfWithWordCounts = dfWithSentences.withColumn(
    "sentence_word_counts",
    transform(
        col("sentences"),
        sentence => size(split(trim(sentence), "\\W+"))
    )
)

val dfExploded = dfWithWordCounts.select(
    col("comment"),
    explode(arrays_zip(col("sentences"), col("sentence_word_counts"))).alias("sentenceData")
).select(
    col("comment"),
    col("sentenceData.sentences").alias("sentence"),
    col("sentenceData.sentence_word_counts").alias("wordCount")
)

dfExploded.show(false)

Key Fixes & Edge Cases to Consider #

• Abbreviation Safety: The regex (?<=[.!?])\\s+ ensures we don't split inside abbreviations like Mr.Smith or U.S.A..
• Empty Sentence Handling: trim() before splitting prevents empty strings from being counted as valid sentences (e.g., if a comment ends with a space after a period).
• Custom Word Definitions: If you need to count only alphabetic words (exclude numbers/punctuation), adjust the split regex and add a filter:

  lambda sentence: F.size(
      F.filter(F.split(F.trim(sentence), "[^a-zA-Z]+"), lambda x: x != "")
  )
  

• Null Handling: Add F.coalesce if your comment column has null values to avoid errors:

  F.split(F.coalesce(F.col("comment"), F.lit("")), "(?<=[.!?])\\s+")
  

内容的提问来源于stack exchange，提问作者user5801828