Scala：具有可序列化的产品不带参数

Question

我的目标是从 csv 文件中读取数据并将我的 rdd 转换为 scala/spark 中的数据帧。这是我的代码：

package xxx.DataScience.CompensationStudy

import org.apache.spark._
import org.apache.log4j._

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf

import org.apache.spark.sql.Row
import org.apache.spark.sql.functions.{col, udf}
import org.apache.spark.sql.types._

import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext


object CompensationAnalysis {

  case class GetDF(profil_date:String, profil_pays:String, param_tarif2:String, param_tarif3:String, dt_titre:String, dt_langues:String,
    dt_diplomes:String, dt_experience:String, dt_formation:String, dt_outils:String, comp_applications:String, 
    comp_interventions:String, comp_competence:String)

  def main(args: Array[String]) {

    Logger.getLogger("org").setLevel(Level.ERROR)

    val conf = new SparkConf().setAppName("CompensationAnalysis ")
    val sc = new SparkContext(conf)

    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    import sqlContext.implicits._


    val lines = sc.textFile("C:/Users/../Downloads/CompensationStudy.csv").flatMap { l => 


      l.split(",") match {

        case field: Array[String] if field.size > 13 => Some(field(0), field(1), field(2), field(3), field(4), field(5), field(6), field(7), field(8), field(9), field(10), field(11), field(12))

        case field: Array[String] if field.size == 1 => Some((field(0), "default value"))

        case _ => None 
      }


    }

在这个舞台上，我遇到了错误：具有可序列化的产品不带参数

    val summary = lines.collect().map(x => GetDF(x("profil_date"), x("profil_pays"), x("param_tarif2"), x("param_tarif3"), x("dt_titre"), x("dt_langues"), x("dt_diplomes"), x("dt_experience"), x("dt_formation"), x("dt_outils"), x("comp_applications"), x("comp_interventions"), x("comp_competence")))

    val sum_df = summary.toDF()

    df.printSchema


  }

}

这是截图：

请帮忙？

Answer 1

您有几件事需要改进。正如@CyrilleCorpet 指出的那样，导致异常的最紧迫的问题是“模式匹配中的三个不同行返回类型Some[Tuple13]、Some[Tuple2] 和Some[Tuple2] 和None.type 的值。最小上限是然后Option[Product with Serializable] 符合flatMap 的签名（结果应该是Iterable[T]）模一些隐式转换。”

基本上，如果你有Some[Tuple13]、Some[Tuple13] 和None 或 Some[Tuple2]、Some[Tuple2] 和None，你会更好。

此外，由于类型擦除，类型上的模式匹配通常不是一个好主意，而且模式匹配对于您的情况甚至都不是很好。

因此您可以在案例类中设置默认值：

case class GetDF(profile_date: String, 
                 profile_pays: String = "default", 
                 param_tarif2: String = "default", 
                 ...
)

然后在你的 lambda 中：

val tokens = l.split
if (l.length > 13) {
   Some(GetDf(l(0), l(1), l(2)...))
} else if (l.length == 1) {
   Some(GetDf(l(0)))
} else {
   None
}

现在在所有情况下，您都返回Option[GetDF]。您可以flatMap RDD 摆脱所有Nones 并仅保留GetDF 实例。