Как использовать столбец, созданный мной в Spark Join? - неоднозначная ошибка

Question

Я боролся с этим некоторое время в скале, и я не могу найти четкого решения для этого.

У меня есть 2 кадра данных:

val Companies = Seq(
  (8, "Yahoo"),
  (-5, "Google"),
  (12, "Microsoft"),
  (-10, "Uber")
).toDF("movement", "Company")

val LookUpTable = Seq(
  ("B", "Buy"),
  ("S", "Sell")
).toDF("Code", "Description")

Мне нужно создать столбец в Companies , который позволит мне присоединиться к таблице поиска. Это простое case утверждение, которое проверяет, является ли движение отрицательным, затем продает, иначе покупает. Затем мне нужно присоединиться к таблице поиска в этом недавно созданном столбце.

val joined = Companies.as("Companies")
    .withColumn("Code",expr("CASE WHEN movement > 0 THEN 'B' ELSE 'S' END"))
    .join(LookUpTable.as("LookUpTable"), $"LookUpTable.Code" === $"Code", "left_outer")

Тем не менее, я получаю следующую ошибку:

org.apache.spark.sql.AnalysisException: Reference 'Code' is ambiguous, could be: Code, LookUpTable.Code.;
  at org.apache.spark.sql.catalyst.expressions.package$AttributeSeq.resolve(package.scala:259)
  at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolveChildren(LogicalPlan.scala:101)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveReferences$$anonfun$40.apply(Analyzer.scala:888)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveReferences$$anonfun$40.apply(Analyzer.scala:890)
  at org.apache.spark.sql.catalyst.analysis.package$.withPosition(package.scala:53)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveReferences$.org$apache$spark$sql$catalyst$analysis$Analyzer$ResolveReferences$$resolve(Analyzer.scala:887)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveReferences$$anonfun$org$apache$spark$sql$catalyst$analysis$Analyzer$ResolveReferences$$resolve$2.apply(Analyzer.scala:896)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveReferences$$anonfun$org$apache$spark$sql$catalyst$analysis$Analyzer$ResolveReferences$$resolve$2.apply(Analyzer.scala:896)
  at org.apache.spark.sql.catalyst.trees.TreeNode$$anonfun$4.apply(TreeNode.scala:329)
  at org.apache.spark.sql.catalyst.trees.TreeNode.mapProductIterator(TreeNode.scala:187)
  at org.apache.spark.sql.catalyst.trees.TreeNode.mapChildren(TreeNode.scala:327)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveReferences$.org$apache$spark$sql$catalyst$analysis$Analyzer$ResolveReferences$$resolve(Analyzer.scala:896)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveReferences$$anonfun$apply$9$$anonfun$applyOrElse$35.apply(Analyzer.scala:956)
  at org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveReferences$$anonfun$apply$9$$anonfun$applyOrElse$35.apply(Analyzer.scala:956)
  at org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$1.apply(QueryPlan.scala:105)
  at org.apache.spark.sql.catalyst.plans.QueryPlan$$anonfun$1.apply(QueryPlan.scala:105

Я пытался добавить псевдоним для кода, но это не работает:

val joined = Companies.as("Companies")
    .withColumn("Code",expr("CASE WHEN movement > 0 THEN 'B' ELSE 'S' END"))
    .join(LookUpTable.as("LookUpTable"), $"LookUpTable.Code" === $"Companies.Code", "left_outer")

org.apache.spark.sql.AnalysisException: cannot resolve '`Companies.Code`' given input columns: [Code, LookUpTable.Code, LookUpTable.Description, Companies.Company, Companies.movement];;
'Join LeftOuter, (Code#102625 = 'Companies.Code)
:- Project [movement#102616, Company#102617, CASE WHEN (movement#102616 > 0) THEN B ELSE S END AS Code#102629]
:  +- SubqueryAlias `Companies`
:     +- Project [_1#102613 AS movement#102616, _2#102614 AS Company#102617]
:        +- LocalRelation [_1#102613, _2#102614]
+- SubqueryAlias `LookUpTable`
   +- Project [_1#102622 AS Code#102625, _2#102623 AS Description#102626]
      +- LocalRelation [_1#102622, _2#102623]

Единственный обходной путь, который я нашел, состоял в том, чтобы создать псевдоним только что созданного столбца, однако затем создается дополнительный столбец, который кажется некорректным.

val joined = Companies.as("Companies")
    .withColumn("_Code",expr("CASE WHEN movement > 0 THEN 'B' ELSE 'S' END")).as("Code")
    .join(LookUpTable.as("LookUpTable"), $"LookUpTable.Code" === $"Code", "left_outer")


joined.show()

+--------+---------+-----+----+-----------+
|movement|  Company|_Code|Code|Description|
+--------+---------+-----+----+-----------+
|       8|    Yahoo|    B|   B|        Buy|
|       8|    Yahoo|    B|   S|       Sell|
|      -5|   Google|    S|   B|        Buy|
|      -5|   Google|    S|   S|       Sell|
|      12|Microsoft|    B|   B|        Buy|
|      12|Microsoft|    B|   S|       Sell|
|     -10|     Uber|    S|   B|        Buy|
|     -10|     Uber|    S|   S|       Sell|
+--------+---------+-----+----+-----------+

Можно ли присоединиться к вновь созданному столбцу, не создавая новый фрейм данных или новый столбец с помощью псевдонима?

Answer 1

пытались ли вы использовать Seq в кадре данных Spark.

1.Использование Seq без повторяющегося столбца

val joined = Companies.as("Companies")
    .withColumn("Code",expr("CASE WHEN movement > 0 THEN 'B' ELSE 'S' END"))
    .join(LookUpTable.as("LookUpTable"), Seq("Code"), "left_outer")

псевдоним после withColumn, но он будет генерировать дубликат столбца

val joined = Companies.withColumn("Code",expr("CASE WHEN movement > 0 THEN 'B' ELSE 'S' END")).as("Companies")
.join(LookUpTable.as("LookUpTable"), $"LookUpTable.Code" === $"Companies.Code", "left_outer")

Answer 2

Выражение можно использовать для объединения:

val codeExpression = expr("CASE WHEN movement > 0 THEN 'B' ELSE 'S' END")
val joined = Companies.as("Companies")
  .join(LookUpTable.as("LookUpTable"), $"LookUpTable.Code" === codeExpression, "left_outer")

Answer 3

Псевдоним потребуется, если вам нужны столбцы из two different dataframes having same name. Это связано с тем, что Spark API DataFrame создает схему для указанного кадра данных, и в данной схеме никогда не может быть двух или более столбцов с одинаковым именем.

Это также причина того, что в SQL,SELECT запрос без псевдонимов работает, но если вы сделаете CREATE TABLE AS SELECT, он выдаст ошибку вроде - duplicate columns.