object not serializable (class: org.apache.kafka.clients.consumer.ConsumerRecord)

3. object not serializable (class: org.apache.kafka.clients.consumer.ConsumerRecord)

 

    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //click^1503305255772^00000003^google^d4d51588-52b3-4bb5-b8b3-a4989ad52872^windows^10.155.167.33

    val region = stream.map(p => p.value())
      .flatMap(t => t.split("\^")).print()