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

Question

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;

此外，特别是对于它得到的短语字段（检查case class[2])

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

如果我将case类[2]中的所有字段都定义为String类型，那么一切都很好，但这不是我想要的。有没有一个简单的方法做这件事[3]？

参考文献

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

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级别上定义列，然后将事情转换为Dataset（比如here或here或here），但我几乎可以肯定，这不是应该做的事情。我也很确定编码器可能是答案，但我不知道如何

Answer 1

TL；使用CSV的DR和标准DataFrame操作进行输入转换是可行的。如果您想避免，您应该使用具有表达性的输入格式（Parquet甚至JSON）。

通常，要转换为静态类型数据集的数据必须已经具有正确的类型。最有效的方法是为CSV阅读器提供模式参数：

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

其中架构可以通过反射来推断：

import org.apache.spark.sql.catalyst.ScalaReflection
 import org.apache.spark.sql.types.StructType

val schema = ScalaReflection.schemaFor[T].dataType.asInstanceOf[StructType]

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]