在 Scala Spark 中加入不同 Dataframe 时动态选择多个列

我有两个 Spark 数据框df1 and df2。有没有办法在连接这两个数据帧时动态选择输出列？以下定义在内部联接的情况下输出 df1 和 df2 中的所有列。

def joinDF (df1: DataFrame,  df2: DataFrame , joinExprs: Column, joinType: String): DataFrame = {   
  val dfJoinResult = df1.join(df2, joinExprs, joinType)
  dfJoinResult
  //.select()
}

输入数据：

val df1 = List(("1","new","current"), ("2","closed","saving"), ("3","blocked","credit")).toDF("id","type","account")
val df2 = List(("1","7"), ("2","5"), ("5","8")).toDF("id","value")

预期结果：

val dfJoinResult = df1
  .join(df2, df1("id") === df2("id"), "inner")
  .select(df1("type"), df1("account"), df2("value")) 

dfJoinResult.schema():

StructType(StructField(type,StringType,true), 
StructField(account,StringType,true), 
StructField(value,StringType,true))

我看过类似的选项df.select(cols.head, cols.tail: _*)但它不允许从两个 DF 中选择列。 有没有办法通过selectExpr动态列以及我们想要从中选择的数据框详细信息def？我正在使用 Spark 2.2.0。

可以通过select表达式为Seq[Column]到方法：

def joinDF(df1: DataFrame,  df2: DataFrame , joinExpr: Column, joinType: String, selectExpr: Seq[Column]): DataFrame = {   
  val dfJoinResult = df1.join(df2, joinExpr, joinType)
  dfJoinResult.select(selectExpr:_*)
}

要调用该方法，请使用：

val joinExpr = df1.col("id") === df2.col("id")
val selectExpr = Seq(df1.col("type"), df1.col("account"), df2.col("value"))

val testDf = joinDF(df1, df2, joinExpr, "inner", selectExpr)

这将给出所需的结果：

+------+-------+-----+
|  type|account|value|
+------+-------+-----+
|   new|current|    7|
|closed| saving|    5|
+------+-------+-----+

In the selectExpr上面，有必要指定列来自哪个数据帧。然而，这可以是如果以下假设成立，则进一步简化:

1. 列到join在两个数据框中具有相同的名称
2. 要选择的列具有唯一的名称（其他数据框没有同名的列）

在这种情况下，joinExpr: Column可以改为joinExpr: Seq[String] and selectExpr: Seq[Column] to selectExpr: Seq[String]:

def joinDF(df1: DataFrame,  df2: DataFrame , joinExpr: Seq[String], joinType: String, selectExpr: Seq[String]): DataFrame = {   
  val dfJoinResult = df1.join(df2, joinExpr, joinType)
  dfJoinResult.select(selectExpr.head, selectExpr.tail:_*)
}

现在调用该方法看起来更干净：

val joinExpr = Seq("id")
val selectExpr = Seq("type", "account", "value")

val testDf = joinDF(df1, df2, joinExpr, "inner", selectExpr)

Note： 当。。。的时候join是使用执行Seq[String]与使用表达式相比，生成的数据帧的列名称将有所不同。当存在同名列时，之后将无法单独选择这些列。