title: Twitter Spark Stream Filtering: Task not serializable exception
body: <p>I am trying to build a Spark application which will filter Twitter streaming feeds based on the list of words I have. I have around 8000 words in my list (Twitter filtering API supports only max upto 400 words). I want to tokenize each incoming tweet into words and then check whether this word exists in my list or not. If any of the word from tweet is in the list, then I should print the tweet, else reject it.</p>
  
  <p>I have written following code to achieve this (I copy paste the following code a few lines at a time on <em>spark-shell</em>, is this the right way to test/run my code?):    </p>
  
  <pre><code>// excluding imports to keep in concise 
  val consumerKey = "" // removed while posting on SOF
  val consumerSecret = "" // removed while posting on SOF
  val accessToken = "" // removed while posting on SOF
  val accessTokenSecret = "" // removed while posting on SOF
  val url = "https://stream.twitter.com/1.1/statuses/filter.json"
  
  val sparkConf = new SparkConf().setAppName("Twitter Sentiment Analysis")
  val sc = new SparkContext(sparkConf)
  
  
  val csvFilterWordsList = sc.textFile("&lt;path to file&gt;/uniq_list_8.0_sorted")
  var filterWordsList : Set[String] = Set()
  for(filterWords &lt;- csvFilterWordsList.collect()) {
      filterWordsList += filterWords.split(",")(0)
  }
  
  
  
  // Twitter Streaming
  val ssc = new JavaStreamingContext(sc,Seconds(2))
  
  val conf = new ConfigurationBuilder()
  conf.setOAuthAccessToken(accessToken)
  conf.setOAuthAccessTokenSecret(accessTokenSecret)
  conf.setOAuthConsumerKey(consumerKey)
  conf.setOAuthConsumerSecret(consumerSecret)
  conf.setStreamBaseURL(url)
  conf.setSiteStreamBaseURL(url)
  
  val filter = Array("twitter")
  
  val auth = AuthorizationFactory.getInstance(conf.build())
  val tweets : JavaReceiverInputDStream[twitter4j.Status] = TwitterUtils.createStream(ssc, auth, filter)
  
  object test extends Serializable {
  def similarity(tweet : twitter4j.Status, wordsList : Set[String]) : String = {
      val tweetTokenized = tweet.getText.replaceAll("[^a-zA-Z0-9]", " ").split(" ")
      var flag = false
      for(word &lt;- tweetTokenized) {
          if(wordsList.contains(word)) {
              flag = true
          }
      }
      if(flag &amp;&amp; tweet.getUser.getLang == "en") {
       return (tweet.getText + "," + tweet.getUser.getLang)
      }
      else {
       return ""
      }
  }
  }
  val statuses = tweets.dstream.map(status =&gt; test.similarity(status,filterWordsList))
  
  statuses.print()
  ssc.start()
  </code></pre>
  
  <p>But on running this I am getting following exception:</p>
  
  <pre><code>org.apache.spark.SparkException: Task not serializable
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166)
      at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158)
      at org.apache.spark.SparkContext.clean(SparkContext.scala:1242)
      at org.apache.spark.streaming.dstream.DStream.map(DStream.scala:436)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:59)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:64)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:66)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:68)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:70)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:72)
      at $iwC$$iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:74)
      at $iwC$$iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:76)
      at $iwC$$iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:78)
      at $iwC$$iwC$$iwC.&lt;init&gt;(&lt;console&gt;:80)
      at $iwC$$iwC.&lt;init&gt;(&lt;console&gt;:82)
      at $iwC.&lt;init&gt;(&lt;console&gt;:84)
      at &lt;init&gt;(&lt;console&gt;:86)
      at .&lt;init&gt;(&lt;console&gt;:90)
      at .&lt;clinit&gt;(&lt;console&gt;)
      at .&lt;init&gt;(&lt;console&gt;:7)
      at .&lt;clinit&gt;(&lt;console&gt;)
      at $print(&lt;console&gt;)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:606)
      at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:789)
      at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1062)
      at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:615)
      at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:646)
      at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:610)
      at org.apache.spark.repl.SparkILoop.reallyInterpret$1(SparkILoop.scala:814)
      at org.apache.spark.repl.SparkILoop.interpretStartingWith(SparkILoop.scala:859)
      at org.apache.spark.repl.SparkILoop.command(SparkILoop.scala:771)
      at org.apache.spark.repl.SparkILoop.processLine$1(SparkILoop.scala:616)
      at org.apache.spark.repl.SparkILoop.innerLoop$1(SparkILoop.scala:624)
      at org.apache.spark.repl.SparkILoop.loop(SparkILoop.scala:629)
      at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply$mcZ$sp(SparkILoop.scala:954)
      at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply(SparkILoop.scala:902)
      at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply(SparkILoop.scala:902)
      at scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
      at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:902)
      at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:997)
      at org.apache.spark.repl.Main$.main(Main.scala:31)
      at org.apache.spark.repl.Main.main(Main.scala)
      at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
      at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
      at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
      at java.lang.reflect.Method.invoke(Method.java:606)
      at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:328)
      at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75)
      at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
  Caused by: java.io.NotSerializableException: twitter4j.conf.ConfigurationBuilder
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1183)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
      at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547)
      at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508)
      at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431)
      at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177)
      at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347)
      at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:42)
      at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:73)
      at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164)
  </code></pre>
  
  <p>But when I just do a simple mapping like<br>
  <code>val statuses = tweets.dstream.map(status =&gt; status.getText)</code><br>
  it works fine.</p>
  
  <p>Can someone please help me on where I am doing something wrong?</p>