Spark 1.5.2: org.apache.spark.sql.AnalysisException: 未解析的运算符'Union'

Question

我有两个数据帧df1和df2。它们都有以下模式：

 |-- ts: long (nullable = true)
 |-- id: integer (nullable = true)
 |-- managers: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- projects: array (nullable = true)
 |    |-- element: string (containsNull = true)

df1是从一个Avro文件创建的，而df2是从一个等效的Parquet文件创建的。然而，如果我执行df1.unionAll(df2).show()，我会得到以下错误：

    org.apache.spark.sql.AnalysisException: unresolved operator 'Union;
    at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$class.failAnalysis(CheckAnalysis.scala:37)
    at org.apache.spark.sql.catalyst.analysis.Analyzer.failAnalysis(Analyzer.scala:44)
    at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:174)
    at org.apache.spark.sql.catalyst.analysis.CheckAnalysis$$anonfun$checkAnalysis$1.apply(CheckAnalysis.scala:49)
    at org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala:103)

- Neel

Answer 1

我遇到了相同的情况，结果发现不仅需要字段相同，还需要在两个数据框中保持字段的完全相同顺序才能使其正常工作。

- Even Cheng

---

是的，顺序很重要！ - soulmachine

数据字段的数据类型也需要考虑。 - David Batista

@Even Cheng，有什么办法可以修复这个问题吗？https://stackoverflow.com/questions/72963925/org-apache-spark-sql-analysisexception-unresolved-operator-project - Shasu

Answer 2

虽然这个问题已经存在了一段时间并且已经有一些答案了，但是我在尝试像合并两个数据框时遇到了这个问题，就像下面的代码：

//Join 2 dataframes
val df = left.unionAll(right)

正如其他人提到的，顺序很重要。因此，请按照与左侧数据框列相同的顺序仅选择正确的列。

//Join 2 dataframes, but take columns in the same order    
val df = left.unionAll(right.select(left.columns.map(col):_*))

- David Royo

---

有什么办法可以修复这个问题吗？https://stackoverflow.com/questions/72963925/org-apache-spark-sql-analysisexception-unresolved-operator-project - Shasu

Answer 3

我在github上发现了以下PR：https://github.com/apache/spark/pull/11333。

这个PR涉及到UDF（用户定义函数）列，在union期间没有被正确处理，从而导致union失败。该PR对此进行了修复，但尚未添加到spark 1.6.2中。我还没有检查spark 2.x的情况。

如果您卡在1.6.x上，有一个愚蠢的解决方法，将DataFrame映射到RDD，再映射回DataFrame。

// for a DF with 2 columns (Long, Array[Long])
val simple = dfWithUDFColumn
  .map{ r => (r.getLong(0), r.getAs[Array[Long]](1))} // DF --> RDD[(Long, Array[Long])]
  .toDF("id", "tags") // RDD --> back to DF but now without UDF column

// dfOrigin has the same structure but no UDF columns
val joined = dfOrigin.unionAll(simple).dropDuplicates(Seq("id")).cache()

- Matti Lyra

---

有什么线索可以修复它吗？https://stackoverflow.com/questions/72963925/org-apache-spark-sql-analysisexception-unresolved-operator-project - Shasu