apache-spark - 这是 Spark 1.3 中的回归错误吗？

Question

在 spark SQL 1.2.1 中没有弃用警告，以下代码在 1.3 中停止工作

工作于 1.2.1 (没有任何弃用警告)

 val sqlContext = new HiveContext(sc)
 import sqlContext._
 val jsonRDD = sqlContext.jsonFile(jsonFilePath)
 jsonRDD.registerTempTable("jsonTable")

 val jsonResult = sql(s"select * from jsonTable")
 val foo = jsonResult.zipWithUniqueId().map {
   case (Row(...), uniqueId) => // do something useful
   ...
 }

 foo.registerTempTable("...")

在 1.3.0 中停止工作(根本无法编译，我所做的只是更改为 1.3)

jsonResult.zipWithUniqueId() //since RDDApi doesn't implement that method

无效的解决方法:

虽然这可能会给我一个 RDD[Row]:

jsonResult.rdd.zipWithUniqueId()  

现在这不能作为 RDD[Row]没有 registerTempTable当然方法

     foo.registerTempTable("...")

这是我的问题

有解决方法吗？ (例如，我只是做错了吗？)

这是一个错误吗？ (我认为在以前的版本中停止编译但没有@deprecated 警告的任何东西显然是一个回归错误)

Answer 1

这不是一个错误，但很抱歉造成困惑!在 Spark 1.3 之前，Spark SQL 被标记为 Alpha 组件，因为 API 仍在不断变化。在 Spark 1.3 中，我们毕业并稳定了 API。在 the documentation 中可以找到关于移植时需要执行的操作的完整说明。 .

我还可以回答您的具体问题，并就我们进行这些更改的原因给出一些理由

Stopped working in 1.3.0 (simply does not compile, and all I did was change to 1.3) jsonResult.zipWithUniqueId() //since RDDApi doesn't implement that method

DataFrames 现在是跨 Scala 和 Java 的统一接口(interface)。然而，由于我们必须在 1.X 的其余部分保持与现有 RDD API 的兼容性， DataFrames不是 RDD s。要获得 RDD 表示，您可以调用 df.rdd或 df.javaRDD
此外，因为我们担心隐式转换可能会发生一些困惑，所以我们规定您必须显式调用 rdd.toDF导致发生从 RDD 的转换。但是，只有当您的 RDD 包含继承自 Product 的对象时，此转换才会自动进行。 (即元组或案例类)。

回到最初的问题，如果你想对具有任意模式的行进行转换，你需要在 map 操作之后明确告诉 Spark SQL 数据的结构(因为编译器不能)。

import org.apache.spark.sql.types._
val jsonData = sqlContext.jsonRDD(sc.parallelize("""{"name": "Michael", "zip": 94709}""" :: Nil))
val newSchema = 
  StructType(
    StructField("uniqueId", IntegerType) +: jsonData.schema.fields)

val augmentedRows = jsonData.rdd.zipWithUniqueId.map { 
  case (row, id) =>
    Row.fromSeq(id +: row.toSeq)
}

val newDF = sqlContext.createDataFrame(augmentedRows, newSchema)