Spark SQL执行flatMap后无法运行group by等操作的问题求助

触发空指针的核心原因是Shuffle阶段遇到非法空值，select * 不走Shuffle、直接读取分区数据返回所以无异常，分组聚合需要跨节点Shuffle数据，处理空值时触发报错，常见的空值来源有2种：

• 原始数据的Description字段存在null值、空字符串，flatMap拆分时没有做前置校验，生成了值为null的word字段
• 构造新DataFrame时指定的字段类型和实际返回值不匹配，比如声明lit为非空整型但实际返回了null值，Shuffle序列化时触发异常

方案1：调整自定义flatMap逻辑 #

1. 先过滤原始数据中的无效Description行
2. 拆分单词时过滤空单词、全空格的无效值
   以下是Scala示例代码：

import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types.StringType

// 第一步：过滤无效原始数据
val cleanDF = originalDF.filter(
  col("Description").isNotNull && 
  col("Description").cast(StringType).trim =!= ""
)

// 第二步：flatMap拆分时增加空校验
val wordDF = cleanDF.flatMap(row => {
  val desc = row.getAs[String]("Description")
  // 按任意空白字符拆分，过滤空单词
  desc.split("\\s+").filter(_.trim.nonEmpty).map(validWord => {
    (row.getAs[Int]("number"), validWord.trim, 1)
  })
}).toDF("number", "word", "lit")

// 第三步：提前校验空值，确认无问题再注册视图
wordDF.filter(col("word").isNull || col("lit").isNull).show()
wordDF.createOrReplaceTempView("data")

如果是PySpark环境，对应代码如下：

from pyspark.sql import Row
from pyspark.sql.functions import col

# 过滤无效原始数据
clean_df = original_df.filter(col("Description").isNotNull() & (col("Description").trim() != ""))

# 自定义flatMap逻辑
def split_desc(row):
    desc = row["Description"]
    words = desc.split()
    valid_words = [w.strip() for w in words if w.strip()]
    return [Row(number=row["number"], word=w, lit=1) for w in valid_words]

word_rdd = clean_df.rdd.flatMap(split_desc)
word_df = spark.createDataFrame(word_rdd)
word_df.createOrReplaceTempView("data")

方案2：使用Spark内置函数实现（更推荐） #

直接用内置的split+explode函数替代自定义flatMap，性能更高且原生处理空值逻辑更稳定，不会出现自定义算子的空指针问题：

import org.apache.spark.sql.functions.{col, split, explode, lit}

val wordDF = originalDF.filter(
  col("Description").isNotNull && 
  col("Description").trim =!= ""
).withColumn("word", explode(split(col("Description"), "\\s+")))
 .filter(col("word").trim =!= "")
 .select("number", "word", lit(1).alias("lit"))

wordDF.createOrReplaceTempView("data")

调整完成后执行分组统计语句即可正常运行：

select word, count(lit) as word_cnt
from data
group by word
order by word_cnt desc

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