full_join将更容易实现，即使您列列表会随时间而变化。
下面是一个使用您的示例的示例

id_cols = ['c1', 'c2']
sum_cols = ['w1', 'w2']

data1_sdf.alias('a'). \
    join(data2_sdf.alias('b'), id_cols, 'full'). \
    selectExpr(*id_cols,
               *['(coalesce(a.{0}, 0) + coalesce(b.{0}, 0)) as {0}'.format(c) for c in sum_cols]
               ). \
    show()

# +---+---+---+---+
# | c1| c2| w1| w2|
# +---+---+---+---+
# |  A|  A|  7| 10|
# |  A|  C|  4|  8|
# |  B|  B|  8|  8|
# |  B|  A| 11|  4|
# |  B|  D|  3|  5|
# |  A|  B|  6|  7|
# +---+---+---+---+

将两个 Dataframe 连接为“full”，并对非空的列求和：

dfa = spark.createDataFrame(data=[["A","A","5","3"],["A","B","6","7"],["B","A","2","3"],["B","B","8","8"]], schema=["C1","C2","W1","W2"])

dfb = spark.createDataFrame(data=[["A","A","2","7"],["A","C","4","8"],["B","A","9","1"],["B","D","3","5"]], schema=["C1","C2","W1","W2"])

result_df = dfa \
  .join(dfb, (dfa.C1 == dfb.C1) & (dfa.C2 == dfb.C2), how="full") \
  .withColumn("C1_new", F.coalesce(dfa.C1, dfb.C1)) \
  .withColumn("C2_new", F.coalesce(dfa.C2, dfb.C2)) \
  .withColumn("W1_sum", F.when(dfa.W1.isNull(), dfb.W1).when(dfb.W1.isNull(), dfa.W1).otherwise((dfa.W1 + dfb.W1).cast('int'))) \
  .withColumn("W2_sum", F.when(dfa.W2.isNull(), dfb.W2).when(dfb.W2.isNull(), dfa.W2).otherwise((dfa.W2 + dfb.W2).cast('int'))) \
  .drop("C1", "C2", "W1", "W2")

[Out]:
+------+------+------+------+
|C1_new|C2_new|W1_sum|W2_sum|
+------+------+------+------+
|     A|     A|     7|    10|
|     A|     B|     6|     7|
|     A|     C|     4|     8|
|     B|     A|    11|     4|
|     B|     B|     8|     8|
|     B|     D|     3|     5|
+------+------+------+------+