如何在Python中高效处理大数据的n-grams统计分析

Gensim 是否适用？ #

Gensim的Phrases模块确实能生成N-gram，但它的核心定位是词向量预处理，对分组统计、去重计数这类业务需求的支持不够直接。如果一定要用Gensim，需要额外配合pandas/Dask完成后续的分组和去重统计，但整体效率未必是最优选择——更推荐结合快速文本分词工具+大数据处理库的组合。

高效实现方案 #

600万条记录的瓶颈主要在纯Python循环的低效和内存占用，核心优化思路是：先按ID聚合去重（同一ID内重复N-gram只保留一次），再利用并行/矢量化工具处理，避免重复计算。

方案1：Polars + spaCy（推荐，内存效率拉满） #

Polars是比pandas更快的大数据处理库，支持懒加载和多核并行；spaCy的分词速度远胜NLTK，适合大规模文本处理。

import polars as pl
import spacy
from itertools import chain

# 加载轻量spaCy模型（禁用不需要的组件提速）
nlp = spacy.load("en_core_web_sm", disable=["parser", "ner"])

def gen_ngrams(text, n):
    # 分词，过滤停用词、标点
    tokens = [tok.lower_ for tok in nlp(text) if not tok.is_stop and not tok.is_punct]
    if len(tokens) < n:
        return []
    # 生成N-gram并去重
    return list({f"{' '.join(tokens[i:i+n])}" for i in range(len(tokens)-n+1)})

# 读取数据（Polars自动懒加载，无需全量入内存）
df = pl.read_csv("your_dataset.csv")

# 步骤1：按ID聚合，生成每个ID的唯一bigram+trigram
id_ngrams = df.group_by("ID").agg(
    pl.col("TEXT").apply(lambda texts: chain.from_iterable(
        gen_ngrams(t, 2) + gen_ngrams(t, 3) for t in texts
    )).alias("all_ngrams")
).with_columns(
    pl.col("all_ngrams").apply(set).alias("unique_ngrams")
)

# 步骤2：关联DATE，按日期分组统计唯一ID数
final_result = df.select(["ID", "DATE"]).unique() \
    .join(id_ngrams, on="ID") \
    .explode("unique_ngrams") \
    .with_columns(
        pl.col("unique_ngrams").str.starts_with("extra").alias("is_extra_start"),
        pl.col("unique_ngrams").str.split(" ").apply(len).alias("ngram_type")
    ) \
    .group_by(["DATE", "unique_ngrams", "is_extra_start", "ngram_type"]) \
    .agg(pl.count("ID").alias("unique_id_count"))

# 查看结果
print(final_result.head())

方案2：Dask + NLTK（兼容现有工具链） #

如果坚持用NLTK，用Dask分块并行处理可以大幅提速：

import dask.dataframe as dd
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
import string
from itertools import chain

stop_words = set(stopwords.words("english") + list(string.punctuation))

def gen_ngrams(text, n):
    tokens = [tok.lower() for tok in word_tokenize(text) if tok.lower() not in stop_words]
    if len(tokens) < n:
        return []
    return list({f"{' '.join(tokens[i:i+n])}" for i in range(len(tokens)-n+1)})

# 读取为Dask DataFrame（自动分块）
ddf = dd.read_csv("your_dataset.csv")

# 按ID聚合生成唯一N-gram
id_ngrams = ddf.groupby("ID").apply(
    lambda x: list(set(chain.from_iterable(
        gen_ngrams(t, 2) + gen_ngrams(t, 3) for t in x["TEXT"]
    ))),
    meta=object
).to_frame("unique_ngrams")

# 关联DATE并统计
final_result = ddf[["ID", "DATE"]].drop_duplicates() \
    .merge(id_ngrams, on="ID") \
    .explode("unique_ngrams") \
    .assign(
        is_extra_start=lambda x: x["unique_ngrams"].str.startswith("extra"),
        ngram_type=lambda x: x["unique_ngrams"].str.split().str.len()
    ) \
    .groupby(["DATE", "unique_ngrams", "is_extra_start", "ngram_type"]) \
    .count() \
    .compute()

print(final_result.head())

方案3：R quanteda + data.table（复用熟悉工具） #

既然你熟悉quanteda，直接用它结合data.table处理效率也很高：

library(quanteda)
library(data.table)

# 高效读取数据
dt <- fread("your_dataset.csv")

# 按ID构建语料库
corpus_id <- corpus(dt, docid_field = "ID", text_field = "TEXT")

# 生成N-gram特征矩阵（自动去重）
toks <- tokens(corpus_id, remove_punct = TRUE, remove_stopwords = TRUE)
toks_ngrams <- tokens_ngrams(toks, n = 2:3)
dfm_ngrams <- dfm(toks_ngrams)

# 转换为data.table并关联DATE
dt_ngrams <- convert(dfm_ngrams, to = "data.table")
dt_ngrams <- melt(dt_ngrams, id.vars = "doc_id", variable.name = "ngram", value.name = "count")
dt_ngrams <- dt_ngrams[count > 0, .(doc_id, ngram)]

# 分组统计唯一ID数
final_result <- unique(dt[, .(ID, DATE)])[dt_ngrams, on = c(ID = "doc_id")][
    , .(unique_id_count = uniqueN(ID)),
    by = .(DATE, ngram, is_extra_start = startsWith(ngram, "extra"), ngram_type = lengths(strsplit(ngram, " ")))
]

head(final_result)

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