如何基于另外两个DataFrame的列向DataFrame添加新列

我明白你想要把df_node_labels里的标签信息（不管是srcLabel还是dstLabel）对应到df_data的每个nodeId上，生成一个统一的label列。直接分两次join会多出冗余列，处理起来麻烦，更简洁的方式是先把所有节点的标签映射整合好，再一次性关联。

步骤1：先把测试数据用Scala代码落地（方便复现） #

先把你给出的两个DataFrame用Spark Scala代码定义出来：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.builder().appName("LabelJoinDemo").master("local[*]").getOrCreate()
import spark.implicits._

// 你的df_data
val df_data = Seq(
  (1, "abc"),
  (2, "def"),
  (3, "fed"),
  (4, "kfl")
).toDF("nodeId", "field1")

// 你的df_node_labels
val df_node_labels = Seq(
  (1, "AAA", 2, "BBB"),
  (2, "BBB", 4, "FFF"),
  (4, "FFF", 3, "CCC")
).toDF("srcId", "srcLabel", "dstId", "dstLabel")

步骤2：整合所有节点的标签映射 #

把df_node_labels里的srcId-srcLabel和dstId-dstLabel两组映射提取出来，合并后去重（避免同一个节点出现重复标签）：

// 提取源节点的id-label映射，重命名列名统一格式
val src_label_map = df_node_labels.select($"srcId".alias("nodeId"), $"srcLabel".alias("label"))
// 提取目标节点的id-label映射，同样统一列名
val dst_label_map = df_node_labels.select($"dstId".alias("nodeId"), $"dstLabel".alias("label"))
// 合并两组映射并去重，得到所有节点的唯一标签映射表
val all_label_maps = src_label_map.union(dst_label_map).distinct()

步骤3：关联到df_data生成最终结果 #

用左连接把标签映射表和df_data关联，这样即使某个节点没有标签（你的数据里所有节点都有，不过这种写法更通用），也不会丢失原数据：

val final_result = df_data.join(all_label_maps, Seq("nodeId"), "left")

// 查看结果
final_result.show()

最终输出效果 #

执行后会得到你想要的结果：

+------+------+-----+
|nodeId|field1|label|
+------+------+-----+
|     1|   abc|  AAA|
|     2|   def|  BBB|
|     3|   fed|  CCC|
|     4|   kfl|  FFF|
+------+------+-----+

这种方式比多次join更简洁，也避免了后续合并多标签列的麻烦，同时保证每个节点的标签唯一。

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