Spark SQL 'explode' 命令在 AWS EC2 上失败，但在本地成功

Posted 2023-04-17

【中文标题】Spark SQL \'explode\' 命令在 AWS EC2 上失败，但在本地成功

【英文标题】：Spark SQL 'explode' command failing on AWS EC2 but succeeding locallySpark SQL 'explode' 命令在 AWS EC2 上失败，但在本地成功

【发布时间】：2015-11-05 22:54:14

【问题描述】：

我正在使用 Spark SQL（我提到它在 Spark 中以防影响 SQL 语法 - 我还不够熟悉，无法确定）并且我有一个表，我正在尝试重新构建它。我有一种可以在本地工作的方法，但是当我尝试在 AWS EC2 实例上运行相同的命令时，我收到一个错误报告，指出我有一个“未解决的操作员”

基本上我的数据看起来像：

userId    someString      varA   
   1      "example1"     [0,2,5] 
   2      "example2"     [1,20,5] 

我在 varA 的 sqlContext 中使用了“explode”命令。当我在本地运行时，事情会正确返回，但在 AWS 上它们会失败。

我可以用以下命令重现这个：

val data = List(
  ("1", "example1", Array(0,2,5)), ("2", "example2", Array(1,20,5)))
val distData = sc.parallelize(data)
val distTable = distData.toDF("userId", "someString", "varA")
distTable.registerTempTable("distTable_tmp")
val temp1 = sqlContext.sql("select userId, someString, varA from distTable_tmp")
val temp2 = sqlContext.sql(
  "select userId, someString, explode(varA) as varA from distTable_tmp")

在本地， temp1.show() 和 temp2.​​show() 返回我所期望的，即：

scala> temp1.show()
+------+----------+----------+
|userId|someString|      varA|
+------+----------+----------+
|     1|  example1| [0, 2, 5]|
|     2|  example2|[1, 20, 5]|
+------+----------+----------+

scala> temp2.show()
+------+----------+----+
|userId|someString|varA|
+------+----------+----+
|     1|  example1|   0|
|     1|  example1|   2|
|     1|  example1|   5|
|     2|  example2|   1|
|     2|  example2|  20|
|     2|  example2|   5|
+------+----------+----+

但在 AWS 上 temp1 sqlContext 命令工作正常，但 temp2 失败并显示以下消息：

scala> val temp2 = sqlContext.sql("select userId, someString, explode(varA) as varA from distTable_tmp")
15/11/05 22:46:49 INFO parse.ParseDriver: Parsing command: select userId, someString, explode(varA) as varA from distTable_tmp
15/11/05 22:46:49 INFO parse.ParseDriver: Parse Completed
org.apache.spark.sql.AnalysisException: unresolved operator 'Project [userId#3,someString#4,HiveGenericUdtf#org.apache.hadoop.hive.ql.udf.generic.GenericUDTFExplode(varA#5) AS varA#6];
...

非常感谢。

【问题讨论】：

您能展示一下您是如何创建sqlContext 的吗？还有你用的是什么版本的 Spark？

@zero323 - 我只是在本地通过 './bin/spark-shell' 和 'MASTER=yarn-client /home/hadoop/spark/bin/spark-shell' 打开 Spark Scala shell在 AWS 上 - 除了似乎为我加载的默认 sqlContext 之外，我没有想到要使用任何东西。 Spark 版本是 1.5.1（本地）和 1.3.1（AWS）。

【参考方案1】：

问题的根源在于您在 EC2 上使用的 Spark 版本。 explode 函数已在 Spark 1.4 中引入，因此无法在 1.3.1 上运行。可以像这样使用RDD 和flatMap：

import org.apache.spark.sql.Row
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.StructType, StructField, IntegerType

val rows: RDD[Row] = distTable.rdd.flatMap(
  row => row.getAs[Seq[Int]](2).map(v => Row.fromSeq(row.toSeq :+ v)))
val newSchema = StructType(
  distTable.schema.fields :+ StructField("varA_exploded", IntegerType, true))

sqlContext.createDataFrame(rows, newSchema).show

// userId someString varA                 varA_exploded
// 1      example1   ArrayBuffer(0, 2, 5) 0            
// 1      example1   ArrayBuffer(0, 2, 5) 2            
// 1      example1   ArrayBuffer(0, 2, 5) 5            
// 2      example2   ArrayBuffer(1, 20... 1            
// 2      example2   ArrayBuffer(1, 20... 20           
// 2      example2   ArrayBuffer(1, 20... 5      

但它怀疑这是否值得大惊小怪。