PySpark中DataFrame调用dot方法报无属性错误如何实现点积运算

PySpark DataFrame 与 pandas DataFrame 是完全独立的两套实现，PySpark DataFrame API 原生没有提供dot方法，直接照搬pandas的df1.dot(df2.T)写法会触发AttributeError: 'DataFrame' object has no attribute 'dot'异常。
DataFrame点积本质是矩阵乘法运算，根据数据规模和使用场景，可以选择以下三种实现方案：

方案1：小数据集转pandas实现 #

如果数据集规模小、可以完全加载到单节点内存，直接转pandas完成点积计算后再转回PySpark DataFrame即可，代码逻辑最简单：

# 转pandas执行点积
pd_res = df1.toPandas().dot(df2.toPandas().T)
# 转回PySpark DataFrame
spark_dot_res = spark.createDataFrame(pd_res)

注意：该方案不适合TB级大规模数据，否则会触发Driver节点OOM。

方案2：大规模数据用MLlib分布式矩阵计算（生产推荐） #

大数据量场景下使用PySpark MLlib自带的分布式矩阵API实现，支持集群分布式计算，不会出现单点内存瓶颈：

from pyspark.mllib.linalg.distributed import IndexedRowMatrix, IndexedRow
from pyspark.ml.linalg import Vectors

# 1. 将两个DataFrame转换为分布式索引行矩阵
# 注意：需要根据实际列结构调整向量组装逻辑，这里默认所有数值列参与运算
mat1 = IndexedRowMatrix(
    df1.rdd.zipWithIndex().map(lambda x: IndexedRow(x[1], Vectors.dense(x[0][:])))
)
mat2 = IndexedRowMatrix(
    df2.rdd.zipWithIndex().map(lambda x: IndexedRow(x[1], Vectors.dense(x[0][:])))
)

# 2. 对应pandas中df2.T的转置逻辑，将右矩阵转置后执行乘法
dot_mat_res = mat1.toBlockMatrix().multiply(mat2.toBlockMatrix().transpose())

# 3. 计算结果转换回PySpark DataFrame
spark_dot_res = dot_mat_res.toIndexedRowMatrix().rows.map(
    lambda row: (row.index, ) + tuple(row.vector.toArray())
).toDF()

注意：执行矩阵乘法前需要保证左矩阵列数与右矩阵转置后的行数匹配，否则会抛出计算维度不匹配的错误。

方案3：逐行点积用Spark SQL内置函数实现 #

如果两个DataFrame行、列完全对齐，仅需要逐行计算对应向量的点积，不需要做全矩阵笛卡尔式乘法，可以直接用Spark SQL内置函数实现，无需依赖MLlib组件：

from pyspark.sql import functions as F

# 提前指定参与点积计算的数值列
calc_columns = df1.columns
# 增加唯一行ID保证两个DataFrame行严格对齐，同时对列加别名避免重名
df1_with_id = df1.withColumn("rid", F.monotonically_increasing_id()).select(
    "rid", *[F.col(c).alias(f"d1_{c}") for c in calc_columns]
)
df2_with_id = df2.withColumn("rid", F.monotonically_increasing_id()).select(
    "rid", *[F.col(c).alias(f"d2_{c}") for c in calc_columns]
)

# 关联后逐行计算：对应列相乘再求和得到点积值
join_df = df1_with_id.join(df2_with_id, on="rid")
dot_calc = sum([F.col(f"d1_{c}") * F.col(f"d2_{c}") for c in calc_columns])
spark_dot_res = join_df.withColumn("dot_val", dot_calc).select("rid", "dot_val")

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