使用 Scala 动态选择带参数的列 - Spark

Question

我需要从我要加入的两个表之一中动态选择列。将要选择的列之一的名称传递给变量。这是详细信息。

表名被传递给变量。 join_id 和 join_type 也是如此。

//Creating scala variables for each table
var table_name_a = dbutils.widgets.get("table_name_a")
var table_name_b = dbutils.widgets.get("table_name_b")

//Create scala variable for Join Id
var join_id = dbutils.widgets.get("table_name_b") + "Id"

// Define join type
var join_type = dbutils.widgets.get("join_type")

然后，我加入表格。我想从表 A 中选择所有列，从表 B 中选择两列：无论在上面的参数中传递什么表 B，一列称为“描述”；第二列与表 B 具有相同的名称，例如，如果表 B 的名称是 Employee，我想从表 B 中选择一个名为“Employee”的列。下面的代码从表 A 中选择所有列，从表中选择 Description 列B（别名）。但我仍然需要从表 B 中选择与表同名的另一列。我事先不知道表 B 总共有多少列，也不知道列顺序或它们的名称 - 因为表 B 是作为参数传递的。

// Joining Tables
var df_joined_tables = df_a
                                     .join(df_b,                                               
                                               df_a(join_id)===df_b(join_id),
                                              join_type
                                          ).select($"df_a.*",$"df_b.Description".alias(table_name_b + " Description"))

我的问题是：如何将变量 table_name_b 作为我试图从表 B 中选择的列传递？

我尝试了下面的代码，这显然是错误的，因为在 "$"df_b.table_name_b" 中，"table_name_b" 应该是参数的内容，而不是列本身的名称。

var df_joined_tables = df_a
                                     .join(df_b,                                               
                                               df_a(join_id)===df_b(join_id),
                                              join_type
                                          ).select($"df_a.*",$"df_b.Description".alias(table_name_b + " Description"),$"df_b.table_name_b")

然后我尝试了下面的代码，它给出了错误：“value table_name_b is not a member of org.apache.spark.sql.DataFrame”

var df_joined_tables = df_a
                                     .join(df_b,                                               
                                               df_a(join_id)===df_b(join_id),
                                              join_type
                                          ).select($"df_a.*",$"df_b.Description".alias(table_name_b + " Description"),df_b.table_name_b)

如何将变量 table_name_b 作为我需要从表 B 中选择的列传递？

Answer 1

您可以构建一个List[org.apache.spark.sql.Column] 并在您的select 函数上使用它，如下例所示：

// sample input:
val df = Seq(
  ("A", 1, 6, 7),
  ("B", 2, 7, 6),
  ("C", 3, 8, 5),
  ("D", 4, 9, 4),
  ("E", 5, 8, 3)
).toDF("name", "col1", "col2", "col3")

df.printSchema()
val columnNames = List("col1", "col2") // string column names from your params
val columnsToSelect = columnNames.map(col(_)) // convert the required column names from string to column type
df.select(columnsToSelect: _*).show() // using the list of columns

// output:
+----+----+
|col1|col2|
+----+----+
|   1|   6|
|   2|   7|
|   3|   8|
|   4|   9|
|   5|   8|
+----+----+

同样可以申请join的

更新

添加另一个例子：

val aliasTableA = "tableA"
val aliasTableB = "tableB"
val joinField = "name"

val df1 = Seq(
  ("A", 1, 6, 7),
  ("B", 2, 7, 6),
  ("C", 3, 8, 5),
  ("D", 4, 9, 4),
  ("E", 5, 8, 3)
).toDF("name", "col1", "col2", "col3")

val df2 = Seq(
  ("A", 11, 61, 71),
  ("B", 21, 71, 61),
  ("C", 31, 81, 51)
).toDF("name", "col_1", "col_2", "col_3")


df1.alias(aliasTableA)
  .join(df2.alias(aliasTableB), Seq(joinField))
  .selectExpr(s"${aliasTableA}.*", s"${aliasTableB}.col_1", s"${aliasTableB}.col_2").show()

// output:
+----+----+----+----+-----+-----+
|name|col1|col2|col3|col_1|col_2|
+----+----+----+----+-----+-----+
|   A|   1|   6|   7|   11|   61|
|   B|   2|   7|   6|   21|   71|
|   C|   3|   8|   5|   31|   81|
+----+----+----+----+-----+-----+