共享 HDInsight SPARK SQL Table saveAsTable 不起作用

Question

我想使用 tableau 显示来自 HDInsight SPARK 的数据。我正在关注这个视频，他们描述了如何连接两个系统并公开数据。

目前我的脚本本身非常简单，如下所示：

 /* csvFile is an RDD of lists, each list representing a line in the CSV file */
val csvLines = sc.textFile("wasb://[e-mail]/*/*/*/mydata__000000.csv")
// Define a schema
case class MyData(Timestamp: String, TimezoneOffset: String, SystemGuid: String, TagName: String, NumericValue: Double, StringValue: String)
// Map the values in the .csv file to the schema
val myData = csvLines.map(s => s.split(",")).filter(s => s(0) != "Timestamp").map(
    s => MyData(s(0),
            s(1),
            s(2),
            s(3),
            s(4).toDouble,
            s(5)         
    )
).toDF()
// Register as a temporary table called "processdata"
myData.registerTempTable("test_table")
myData.saveAsTable("test_table") 

不幸的是，我遇到了以下错误

warning: there were 1 deprecation warning(s); re-run with -deprecation for details
org.apache.spark.sql.AnalysisException: Table `test_table` already exists.;
    at org.apache.spark.sql.DataFrameWriter.saveAsTable(DataFrameWriter.scala:209)
    at org.apache.spark.sql.DataFrameWriter.saveAsTable(DataFrameWriter.scala:198)

我还尝试使用以下代码覆盖表（如果存在）

   import org.apache.spark.sql.SaveMode
    myData.saveAsTable("test_table", SaveMode.Overwrite) 

但它仍然给了我同样的错误。

warning: there were 1 deprecation warning(s); re-run with -deprecation for details
java.lang.RuntimeException: Tables created with SQLContext must be TEMPORARY. Use a HiveContext instead.
    at scala.sys.package$.error(package.scala:27)
    at org.apache.spark.sql.execution.SparkStrategies$DDLStrategy$.apply(SparkStrategies.scala:416)
    at org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
    at org.apache.spark.sql.catalyst.planning.QueryPlanner$$anonfun$1.apply(QueryPlanner.scala:58)
    at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)

有人可以帮我解决这个问题吗？

Answer 1

我知道这是我的错误，但我会把它作为一个答案，因为它在任何博客或论坛答案中都不容易获得。 希望它能帮助像我这样的人，从 Spark 开始

我发现.toDF()实际上创建了sqlContext，而不是基于hiveContext的DataFrame。 所以我现在已经更新了我的代码，如下所示

// Map the values in the .csv file to the schema
val myData = csvLines.map(s => s.split(",")).filter(s => s(0) != "Timestamp").map(
    s => MyData(s(0),
            s(1),
            s(2),
            s(3),
            s(4).toDouble,
            s(5)         
    )
)
// Register as a temporary table called "myData"
val myDataFrame = hiveContext.createDataFrame(myData)
myDataFrame.registerTempTable("mydata_stored")
myDataFrame.write.mode(SaveMode.Overwrite).saveAsTable("mydata_stored")

还要确保 s（4） 具有适当的双精度值，否则添加 try/catch 来处理它。 我做了这样的事情：

def parseDouble(s: String): Double = try { s.toDouble } catch { case _ => 0.00 }
parseDouble(s(4))

问候基兰