如何将 csv 直接加载到 Spark 数据集中?

Question

我有一个csv文件[1]，我想直接加载到数据集中。问题是我总是收到这样的错误

org.apache.spark.sql.AnalysisException: Cannot up cast `probability` from string to float as it may truncate
The type path of the target object is:
- field (class: "scala.Float", name: "probability")
- root class: "TFPredictionFormat"
You can either add an explicit cast to the input data or choose a higher precision type of the field in the target object;

此外，特别是对于phrases字段(检查案例类 [2])，它得到

org.apache.spark.sql.AnalysisException: cannot resolve '`phrases`' due to data type mismatch: cannot cast StringType to ArrayType(StringType,true);

如果我将案例类 [2] 中的所有字段定义为字符串类型，那么一切正常，但这不是我想要的。有没有简单的方法可以做到这一点[3]？

---

引用

[1] 示例行

B017NX63A2,Merrell,"['merrell_for_men', 'merrell_mens_shoes', 'merrel']",merrell_shoes,0.0806054356579781

[2] 我的代码片段如下

import spark.implicits._
val INPUT_TF = "<SOME_URI>/my_file.csv"
final case class TFFormat (
doc_id: String,
brand: String,
phrases: Seq[String],
prediction: String,
probability: Float
)
val ds = sqlContext.read
.option("header", "true")
.option("charset", "UTF8")
.csv(INPUT_TF)
.as[TFFormat]
ds.take(1).map(println)

[3] 我已经找到了通过首先在 DataFrame 级别定义列并将内容转换为数据集(如此处或此处或此处)来做到这一点的方法，但我几乎可以肯定这不是应该完成的方式。我也非常确定编码器可能是答案，但我不知道如何

Answer 1

TL;灾难恢复使用csv输入转换与标准DataFrame操作是要走的路。如果你想避免，你应该使用具有表现力的输入格式(Parquet甚至JSON)。

通常，要转换为静态类型数据集的数据必须已经具有正确的类型。最有效的方法是为csv读者提供schema论据：

val schema: StructType = ???
val ds = spark.read
.option("header", "true")
.schema(schema)
.csv(path)
.as[T]

其中schema可以通过反射推断

：

import org.apache.spark.sql.catalyst.ScalaReflection
import org.apache.spark.sql.types.StructType
val schema = ScalaReflection.schemaFor[T].dataType.asInstanceOf[StructType]

不幸的是，它不适用于您的数据和类csv因为阅读器不支持ArrayType(但它适用于像FloatType这样的原子类型)，所以你必须使用艰难的方式。一个幼稚的解决方案可以表示如下：

import org.apache.spark.sql.functions._
val df: DataFrame = ???  // Raw data
df
.withColumn("probability", $"probability".cast("float"))
.withColumn("phrases",
split(regexp_replace($"phrases", "[\['\]]", ""), ","))
.as[TFFormat]

但是您可能需要更复杂的东西，具体取决于phrases的内容。