如果 dfA 对于广播连接来说足够小，那么这应该可以工作：

dfB.join(dfA, how="left", on=F.col("src") == F.col("id")).select(
    "src", "dst", F.coalesce(F.col("value"), F.lit(0)).alias("v1")
).join(dfA, how="left", on=F.col("src") == F.col("id")).select(
    "src", "dst", (F.col("v1") + F.coalesce(F.col("value"), F.lit(0))).alias("sum")
)

您可以删除 .coalesce() ，如果id列包含每个src和dst值。有一些方法可以实现这一点，但你最好的选择可能是使用 .transform() .

def join_sum(join_df):
    def _(df):
        return (
            df.join(join_df, how="left", on=F.col("src") == F.col("id"))
            .select("src", "dst", F.coalesce(F.col("value"), F.lit(0)).alias("v1"))
            .join(join_df, how="left", on=F.col("src") == F.col("id"))
            .select(
                "src",
                "dst",
                (F.col("v1") + F.coalesce(F.col("value"), F.lit(0))).alias("sum"),
            )
        )

    return _

dfB.transform(join_sum(dfA))

基本上你需要加入2个Dataframe的条件 (id = src OR id = dst) 然后分组对列求和 value :

from pyspark.sql import functions as F

output = df_a.join(
    df_b, 
    (F.col("id") == F.col("src")) | (F.col("id") == F.col("dst"))
).groupBy("src", "dst").agg(F.sum("value").alias("sum"))

output.show()

# +---+---+-----+

# |src|dst|  sum|

# +---+---+-----+

# |  2|  3|100.3|

# |  1|  2| 10.3|

# |  3|  1|110.0|

# +---+---+-----+