在Scala Spark中连接不同数据帧时动态选择多列

Question

我有两个火花数据帧DF1和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中从中选择的数据帧详细信息？我使用的是Spark2.2.0。

Answer 1

可以将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|
+------+-------+-----+

null

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)

注意：当使用seq[String]执行join时，与使用表达式相比，结果数据帧的列名将有所不同。当存在具有相同名称的列时，将无法在之后单独选择这些列。