如何不调用toPandas或todense将scipy稀疏矩阵转为PySpark DataFrame？

针对你的大型CSR稀疏矩阵转PySpark DataFrame的需求，推荐利用稀疏矩阵本身的结构特性，结合Spark的分布式计算能力实现高效转换，避免密集化或循环创建小DataFrame的低效操作。

核心思路 #

CSR矩阵的核心由三个数组组成：

• indptr: 行指针数组，标记每行数据在indices和data中的起始/结束位置
• indices: 非零元素的列索引数组
• data: 非零元素的数值数组

我们可以通过广播这三个数组到所有Executor，并行处理每行数据，生成Spark支持的SparseVector类型，最终转换为DataFrame。这种方式无需将矩阵密集化，也不会在驱动端执行大量循环操作。

代码实现 #

from pyspark.ml.linalg import SparseVector

# 获取CSR矩阵的核心属性
n_rows = data_transformed.shape[0]
n_cols = data_transformed.shape[1]
indptr = data_transformed.indptr
indices = data_transformed.indices
data = data_transformed.data

# 将核心数组广播到所有Executor（避免重复传输大数组）
broadcast_indptr = spark.sparkContext.broadcast(indptr)
broadcast_indices = spark.sparkContext.broadcast(indices)
broadcast_data = spark.sparkContext.broadcast(data)

# 定义生成每行稀疏向量的函数
def build_sparse_row(row_idx):
    start_pos = broadcast_indptr.value[row_idx]
    end_pos = broadcast_indptr.value[row_idx + 1]
    # 生成对应行的SparseVector
    return (
        row_idx,
        SparseVector(n_cols, broadcast_indices.value[start_pos:end_pos], broadcast_data.value[start_pos:end_pos])
    )

# 并行处理所有行，转换为RDD
sparse_rows_rdd = spark.sparkContext.parallelize(range(n_rows)).map(build_sparse_row)

# 转换为PySpark DataFrame
df = sparse_rows_rdd.toDF(["row_id", "features"])

关键优势 #

1. 内存效率: 全程无需密集化矩阵，仅操作CSR的三个压缩数组，内存占用极低
2. 分布式处理: 利用Spark的并行计算能力，所有行的处理在Executor分布式执行，避免驱动端瓶颈
3. Schema简洁: 用SparseVector封装高维稀疏特征，避免创建7万多列的超大DataFrame（这种结构在Spark中性能极差且易触发OOM）

关于你之前方法的问题分析 #

• 按行批处理密集化: 转密集数组会瞬间占用巨量内存（10万行×7万列的float64矩阵约占550GB内存），且Spark无法高效推断超大量列的Schema
• 按列循环创建DataFrame: 循环创建7万多个小DataFrame会产生大量序列化/反序列化开销，合并操作的复杂度也会导致效率极低

内容的提问来源于stack exchange，提问作者R. Kulebyakin