如何更改Spark數據集上的模式

Question

當我使用select語句在Spark 2中檢索數據集時，基礎列會繼承查詢列的數據類型。

val ds1 = spark.sql("select 1 as a, 2 as b, 'abd' as c") 

ds1.printSchema() 
root 
|-- a: integer (nullable = false) 
|-- b: integer (nullable = false) 
|-- c: string (nullable = false) 

現在，如果我將它轉換爲case類，它將正確轉換值，但底層架構仍然是錯誤的。

case class abc(a: Double, b: Double, c: String) 
val ds2 = ds1.as[abc] 
ds2.printSchema() 
root 
|-- a: integer (nullable = false) 
|-- b: integer (nullable = false) 
|-- c: string (nullable = false) 

ds2.collect 
res18: Array[abc] = Array(abc(1.0,2.0,abd)) 

我「應該」可以指定編碼器，當我創建第二個數據集，但斯卡拉似乎忽略此參數來使用（這是一個BUG？）：

val abc_enc = org.apache.spark.sql.Encoders.product[abc] 

val ds2 = ds1.as[abc](abc_enc) 

ds2.printSchema 
root 
|-- a: integer (nullable = false) 
|-- b: integer (nullable = false) 
|-- c: string (nullable = false) 

所以只有這樣我才能看到這樣做，簡單地說，沒有非常複雜的映射就是使用createDataset，但是這需要在底層對象上進行收集，所以它並不理想。

val ds2 = spark.createDataset(ds1.as[abc].collect) 

---

Answer 1

您可以簡單地使用上columnscast方法

import sqlContext.implicits._ 
val ds2 = ds1.select($"a".cast(DoubleType), $"a".cast(DoubleType), $"c") 
ds2.printSchema() 

你應該有

root 
|-- a: double (nullable = false) 
|-- a: double (nullable = false) 
|-- c: string (nullable = false) 

Answer 2

你也可以轉換列，同時使用SQL查詢選擇如下

import spark.implicits._ 

val ds = Seq((1,2,"abc"),(1,2,"abc")).toDF("a", "b","c").createOrReplaceTempView("temp") 

val ds1 = spark.sql("select cast(a as Double) , cast (b as Double), c from temp") 

ds1.printSchema() 

這有模式作爲

root 
|-- a: double (nullable = false) 
|-- b: double (nullable = false) 
|-- c: string (nullable = true) 

現在你可以轉換爲數據集與案例類

case class abc(a: Double, b: Double, c: String) 

val ds2 = ds1.as[abc] 
ds2.printSchema() 

現在有需要的架構

root 
|-- a: double (nullable = false) 
|-- b: double (nullable = false) 
|-- c: string (nullable = true) 

希望這有助於！

Answer 3

好的，我想我已經以更好的方式解決了這個問題。

當我們創建一個新的數據集時，我們可以引用數據集的rdd而不是使用collect。

所以不是

val ds2 = spark.createDataset(ds1.as[abc].collect) 

我們使用：

val ds2 = spark.createDataset(ds1.as[abc].rdd) 

ds2.printSchema 
root 
|-- a: double (nullable = false) 
|-- b: double (nullable = false) 
|-- c: string (nullable = true) 

這使懶惰的評估不變，但允許新的數據集採用編碼器爲ABC的情況下類，以及隨後的架構當我們用它來創建一個新表格時會反映這一點。

Answer 4

這是Spark API開放的問題（檢查這張票SPARK-17694）

所以，你需要做的是做一個額外的顯式類型轉換。像這樣的東西應該工作：

ds1.as[abc].map(x => x : abc)