使用PySpark/Pandas查找分组后所有行中共同出现的单词

核心逻辑 #

你需要的「每组内所有行都重复出现的单词」，等价于该单词在对应(person_id, category)分组内的出现次数 = 该分组的总行数，你已经统计出了每个单词的出现次数，只需补充计算每个分组的总行数，过滤匹配后拼接单词即可。

实现方式 #

1. Spark SQL 实现（从原始表直接计算） #

假设你的原始数据表名为source_table，可直接执行如下SQL得到结果：

WITH group_total AS (
    -- 统计每个分组的总行数
    SELECT 
        person_id,
        category,
        COUNT(1) AS row_total
    FROM source_table
    GROUP BY person_id, category
),
word_explode AS (
    -- 拆分notes字段为单个单词，每行对应一个单词
    SELECT 
        person_id,
        category,
        EXPLODE(SPLIT(notes, ' ')) AS word
    FROM source_table
),
word_count AS (
    -- 统计每个分组下各单词的出现次数
    SELECT 
        person_id,
        category,
        word,
        COUNT(1) AS word_cnt
    FROM word_explode
    GROUP BY person_id, category, word
)
-- 过滤符合条件的单词并拼接得到最终结果
SELECT 
    wc.person_id,
    wc.category,
    CONCAT_WS(' ', COLLECT_LIST(wc.word)) AS notes
FROM word_count wc
INNER JOIN group_total gt 
ON wc.person_id = gt.person_id AND wc.category = gt.category
WHERE wc.word_cnt = gt.row_total
GROUP BY wc.person_id, wc.category
ORDER BY wc.person_id, wc.category

2. 基于你已有的中间词频结果处理 #

如果不想重新从原始表计算，你已经得到的中间结果表为mid_table（notes字段格式为He (2) bought(2) cat(1) dog(1)），可按如下逻辑处理：

WITH group_total AS (
    SELECT person_id, category, COUNT(1) AS row_total
    FROM source_table
    GROUP BY person_id, category
),
word_parse AS (
    SELECT 
        person_id,
        category,
        -- 拆分解析出单词和对应次数
        EXPLODE(TRANSFORM(SPLIT(notes, ' '), s -> 
            STRUCT(
                REGEXP_EXTRACT(s, '^([a-zA-Z]+)', 1) AS word,
                CAST(REGEXP_EXTRACT(s, '(\\d+)\\)', 1) AS INT) AS word_cnt
            )
        )) AS word_info
    FROM mid_table
)
SELECT 
    wp.person_id,
    wp.category,
    CONCAT_WS(' ', COLLECT_LIST(wp.word_info.word)) AS notes
FROM word_parse wp
INNER JOIN group_total gt 
ON wp.person_id = gt.person_id AND wp.category = gt.category
WHERE wp.word_info.word_cnt = gt.row_total
GROUP BY wp.person_id, wp.category

3. RDD 实现（Scala 示例） #

// 读取原始文件，过滤表头
val rawRdd = sc.textFile("你的文件路径")
  .filter(line => !line.startsWith("person_id"))
  .map(line => {
    val arr = line.split("\\|").map(_.trim)
    ((arr(0), arr(1)), arr(2))
  })

// 统计每个分组的总行数
val groupCountMap = rawRdd.countByKey()

// 分组处理得到最终结果
val resultRdd = rawRdd.groupByKey()
  .map { case ((pid, cate), notesIter) =>
    val totalRows = groupCountMap((pid, cate))
    // 统计当前组所有单词的出现次数
    val wordCountMap = notesIter.flatMap(_.split(" "))
      .groupBy(identity)
      .mapValues(_.size)
    // 过滤符合条件的单词拼接
    val resultNotes = wordCountMap.filter(_._2 == totalRows).keys.mkString(" ")
    s"$pid|$cate|$resultNotes"
  }

// 输出结果
resultRdd.collect().foreach(println)

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