title: mapPartitions compile error: missing parameter type
body: <p>I'm trying to read a stream from a Kafka source containing JSON records using a pattern from the book <a href="https://github.com/holdenk/learning-spark-examples/blob/master/src/main/scala/com/oreilly/learningsparkexamples/scala/BasicParseJsonWithJackson.scala" rel="nofollow noreferrer">Learning Spark</a>:</p>
  
  <pre class="lang-scala prettyprint-override"><code>import spark.implicits._
  import org.apache.spark.sql.functions._
  import org.apache.spark.sql.types._
  import org.apache.spark.sql.SparkSession
  import java.util.Date
  
  import com.fasterxml.jackson.databind.ObjectMapper
  import com.fasterxml.jackson.module.scala.experimental.ScalaObjectMapper
  import com.fasterxml.jackson.module.scala.DefaultScalaModule
  import com.fasterxml.jackson.databind.DeserializationFeature
  
  val spark = SparkSession.builder.appName("StreamingRetailTransactions").config("master", "local").getOrCreate()
  
  
  val df = spark.readStream.
                  format("kafka").
                  option("kafka.bootstrap.servers", ...).
                  option("subscribe", "transactions_load").
                  option("kafka.security.protocol", "SASL_SSL").
                  ...
                  load()
  
  case class Invoice(
                      invoiceNo: Int,
                      stockCode: Int,
                      description: String,
                      ...
                      storeId: Int,
                      transactionId: String
                    )
  </code></pre>
  
  <p>And then ...</p>
  
  <pre class="lang-scala prettyprint-override"><code>val df2 = df.selectExpr("CAST(value AS String)").as[String]
  
  val df3 = df2.mapPartitions(records =&gt; {
  
    val mapper = new ObjectMapper with ScalaObjectMapper
    mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
    mapper.registerModule(DefaultScalaModule)
  
    records.flatMap(record =&gt; {
      try {
        Some(mapper.readValue(record, classOf[Invoice]))
      } catch {
        case e: Exception =&gt; None
      }
    })
  }, true)
  
  val query = df3.writeStream.format("console").start()
  </code></pre>
  
  <p>However, I'm running into this issue:</p>
  
  <pre class="lang-none prettyprint-override"><code>df2: org.apache.spark.sql.Dataset[String] = [value: string]
  &lt;console&gt;:63: error: missing parameter type
         val df3 = df2.mapPartitions(records =&gt; {
                                     ^
  </code></pre>
  
  <p>Any ideas?</p>