适配16G内存Mac笔记本的Python可扩展数据处理引擎推荐

针对你在16G内存Mac上处理100G以内购物篮数据、统计每对购物篮共同商品数的需求，以下是几个支持磁盘分组/排序的工具及具体实现方案：

1. PySpark（本地模式） #

PySpark的本地模式原生支持磁盘溢出，当内存不足以承载数据时，会自动将分组、排序过程中的数据写入磁盘，完全适配你的硬件条件。

实现思路 #

1. 读取制表符分隔文件，解析为（商品ID，购物篮ID）键值对
2. 按商品ID分组，得到每个商品对应的所有购物篮ID集合
3. 对每个商品的购物篮ID生成两两组合（仅保留篮A < 篮B的组合避免重复统计）
4. 按购物篮对分组，统计每组出现次数（即共同商品数量）
5. 输出结果到指定格式文件

关键代码 #

from pyspark.sql import SparkSession
from itertools import combinations

# 初始化本地模式Spark会话，调整分区数适配CPU核心
spark = SparkSession.builder \
    .master("local[*]") \
    .appName("BasketCooccurrence") \
    .config("spark.sql.shuffle.partitions", "8") \
    .config("spark.driver.memory", "8g")  # 分配8G驱动内存，剩余内存留给系统
    .getOrCreate()

# 读取输入文件
df = spark.read.csv("input.tsv", sep="\t", header=False, inferSchema=True)
df = df.withColumnRenamed("_c0", "basket_id").withColumnRenamed("_c1", "product_id")

# 转换为RDD处理分组与组合
product_baskets = df.rdd.map(lambda x: (x.product_id, x.basket_id)).groupByKey()
basket_pairs = product_baskets.flatMap(lambda x: combinations(sorted(x[1]), 2))
pair_counts = basket_pairs.map(lambda x: ((x[0], x[1]), 1)).reduceByKey(lambda a, b: a + b)

# 转换为DataFrame并输出
result_df = pair_counts.map(lambda x: (x[0][0], x[0][1], x[1])).toDF(["basket_a", "basket_b", "common_products"])
result_df.write.csv("output.tsv", sep="\t", header=False, mode="overwrite")

spark.stop()

2. Dask（本地批处理） #

Dask是轻量级并行计算框架，API贴近Pandas，原生支持分块+磁盘溢出，无需复杂配置即可处理超内存数据，对Mac环境友好。

关键代码 #

import dask.dataframe as dd
from itertools import combinations
import dask.bag as db

# 分块读取输入文件，默认64MB分块可根据内存调整
df = dd.read_csv("input.tsv", sep="\t", header=None, names=["basket_id", "product_id"])

# 按商品ID分组后生成购物篮对
product_baskets = df.groupby("product_id")["basket_id"].apply(list, meta=("basket_id", list)).to_bag()

def generate_pairs(item):
    _, baskets = item
    return combinations(sorted(baskets), 2)

basket_pairs = product_baskets.flatMap(generate_pairs)
pair_counts = basket_pairs.frequencies()

# 转换格式并输出为单文件
result_df = pair_counts.to_dataframe(columns=["basket_pair", "common_products"])
result_df[["basket_a", "basket_b"]] = result_df["basket_pair"].apply(lambda x: (x[0], x[1]), meta=("basket_a", int, "basket_b", int))
result_df = result_df.drop("basket_pair", axis=1)
result_df.to_csv("output_dask.tsv", sep="\t", index=False, single_file=True)

3. Pandas（手动分块处理） #

如果熟悉Pandas，可通过手动分块+磁盘缓存实现，无需额外框架依赖。

关键代码 #

import pandas as pd
from itertools import combinations
import os

# 分块读取并生成中间文件
chunk_size = 10**6  # 每块100万行，可根据内存调整
intermediate_dir = "temp_intermediate"
os.makedirs(intermediate_dir, exist_ok=True)

for idx, chunk in enumerate(pd.read_csv("input.tsv", sep="\t", header=None, names=["basket_id", "product_id"], chunksize=chunk_size)):
    chunk[["product_id", "basket_id"]].to_csv(f"{intermediate_dir}/chunk_{idx}.tsv", sep="\t", index=False, header=False)

# 合并中间文件，按商品ID聚合购物篮
product_baskets = {}
for file in os.listdir(intermediate_dir):
    chunk = pd.read_csv(f"{intermediate_dir}/{file}", sep="\t", header=None, names=["product_id", "basket_id"])
    for pid, baskets in chunk.groupby("product_id")["basket_id"]:
        product_baskets.setdefault(pid, set()).update(baskets)

# 统计购物篮对的共同商品数
pair_counts = {}
for baskets in product_baskets.values():
    for pair in combinations(sorted(baskets), 2):
        pair_counts[pair] = pair_counts.get(pair, 0) + 1

# 输出结果
result_df = pd.DataFrame(pair_counts.items(), columns=["basket_pair", "common_products"])
result_df[["basket_a", "basket_b"]] = pd.DataFrame(result_df["basket_pair"].tolist(), index=result_df.index)
result_df = result_df.drop("basket_pair", axis=1)
result_df.to_csv("output_pandas.tsv", sep="\t", index=False)

# 清理中间文件
for file in os.listdir(intermediate_dir):
    os.remove(f"{intermediate_dir}/{file}")
os.rmdir(intermediate_dir)

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