Spark не может найти кодировщик (case case), хотя и предоставляет его

Question

Попытка выяснить, почему при возникновении ошибки на кодировщиках было бы полезно любое понимание!

ОШИБКА Невозможно найти кодировщик для типа SolrNewsDocument, для хранения `* необходим неявный кодировщик [SolrNewsDocument]1004 *

Очевидно, я импортировал spark.implicits._.Я также предоставил кодировщик в качестве класса case.

def ingestDocsToSolr(newsItemDF: DataFrame) = {
  case class SolrNewsDocument(
                             title: String,
                             body: String,
                             publication: String,
                             date: String,
                             byline: String,
                             length: String
                           )
  import spark.implicits._
  val solrDocs = newsItemDF.as[SolrNewsDocument].map { doc =>
    val solrDoc = new SolrInputDocument
    solrDoc.setField("title", doc.title.toString)
    solrDoc.setField("body", doc.body)
    solrDoc.setField("publication", doc.publication)
    solrDoc.setField("date", doc.date)
    solrDoc.setField("byline", doc.byline)
    solrDoc.setField("length", doc.length)

    solrDoc
  }

  // can be used for stream SolrSupport.
  SolrSupport.indexDocs("localhost:2181", "collection", 10, solrDocs.rdd);
  val solrServer = SolrSupport.getCachedCloudClient("localhost:2181")
  solrServer.setDefaultCollection("collection")
  solrServer.commit(false, false)
}

Answer 1

Я получил эту ошибку, пытаясь перебрать текстовый файл, и в моем случае, как и в версии 2.4.x, проблема заключалась в том, что мне пришлось сначала привести его к RDD (который раньше был неявным)

textFile
  .rdd
  .flatMap(line=>line.split(" "))

Миграция нашей кодовой базы Scala в Spark 2

Answer 2

//Check this one.-Move case class declaration before function declaration.
//Encoder is created once case class statement is executed by compiler. Then only compiler will be able to use encoder inside function deceleration.


import spark.implicits._

case class SolrNewsDocument(title: String,body: String,publication: String,date: String,byline: String,length: String)


def ingestDocsToSolr(newsItemDF:DataFrame) = {
val solrDocs = newsItemDF.as[SolrNewsDocument]}