动态连接多个列上的两个 spark-scala 数据帧,无需硬编码连接条件
dynamically join two spark-scala dataframes on multiple columns without hardcoding join conditions
我想在多个列上动态加入两个 spark-scala 数据帧。我会避免硬编码列名称比较,如以下语句所示;
val joinRes = df1.join(df2, df1("col1") == df2("col1") and df1("col2") == df2("col2"))
pyspark 版本中已存在此查询的解决方案 -- 提供如下 link
我想使用 spark-scala 编写相同的代码
在 scala 中,您可以使用与 python 中类似的方式进行操作,但您需要使用 map 和 reduce 函数:
val sparkSession = SparkSession.builder().getOrCreate()
import sparkSession.implicits._
val df1 = List("a,b", "b,c", "c,d").toDF("col1","col2")
val df2 = List("1,2", "2,c", "3,4").toDF("col1","col2")
val columnsdf1 = df1.columns
val columnsdf2 = df2.columns
val joinExprs = columnsdf1
.zip(columnsdf2)
.map{case (c1, c2) => df1(c1) === df2(c2)}
.reduce(_ && _)
val dfJoinRes = df1.join(df2,joinExprs)
我想在多个列上动态加入两个 spark-scala 数据帧。我会避免硬编码列名称比较,如以下语句所示;
val joinRes = df1.join(df2, df1("col1") == df2("col1") and df1("col2") == df2("col2"))
pyspark 版本中已存在此查询的解决方案 -- 提供如下 link
我想使用 spark-scala 编写相同的代码
在 scala 中,您可以使用与 python 中类似的方式进行操作,但您需要使用 map 和 reduce 函数:
val sparkSession = SparkSession.builder().getOrCreate()
import sparkSession.implicits._
val df1 = List("a,b", "b,c", "c,d").toDF("col1","col2")
val df2 = List("1,2", "2,c", "3,4").toDF("col1","col2")
val columnsdf1 = df1.columns
val columnsdf2 = df2.columns
val joinExprs = columnsdf1
.zip(columnsdf2)
.map{case (c1, c2) => df1(c1) === df2(c2)}
.reduce(_ && _)
val dfJoinRes = df1.join(df2,joinExprs)