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()