I have some DStream in Spark Scala and I want to sort it then take the top N. The problem is that whenever I try to run it I get NotSerializableException and the exception message says:

This is because the DStream object is being referred to from within the closure.

The problem is that I don't know how to solve it.

My try is attached with the e-mail.

I don't mind any other ways to sort a DStream and get its top N rather than my way.