title: Can't pickle _thread.lock objects Pyspark send request to elasticseach
body: <p>I am using pyspark streaming to collect data from tweepy. After all the set up, I send the dict(json) to elasticsearch via elasticsearch.index(). But I get "can't pickle_thread.lock objects" error and other 63 errors. The track back log is too long to show in my console!</p>
  
  <p>The design is that I get a json/dict type file, convert it into an DStream, add another feature names "sentiment" to it by calling TextBlob in a map() function. It all works fine, but when I add another map function to call elasticsearch.index(), I get the error.</p>
  
  <p>Below is the part of the super long error log in my console.</p>
  
  <blockquote>
    <p>Blockquote
    During handling of the above exception, another exception occurred:
    Traceback (most recent call last):
      File "/Users/ayane/anaconda/lib/python3.6/site-packages/pyspark/streaming/util.py", line 105, in dumps
        func.func, func.rdd_wrap_func, func.deserializers)))
      File "/Users/ayane/anaconda/lib/python3.6/site-packages/pyspark/serializers.py", line 460, in dumps
        return cloudpickle.dumps(obj, 2)
      File "/Users/ayane/anaconda/lib/python3.6/site-packages/pyspark/cloudpickle.py", line 704, in dumps
        cp.dump(obj)
      File "/Users/ayane/anaconda/lib/python3.6/site-packages/pyspark/cloudpickle.py", line 162, in dump
        raise pickle.PicklingError(msg)
    _pickle.PicklingError: Could not serialize object: TypeError: can't pickle _thread.lock objects
        at org.apache.spark.streaming.api.python.PythonTransformFunctionSerializer$.serialize(PythonDStream.scala:144)
        at org.apache.spark.streaming.api.python.TransformFunction$$anonfun$writeObject$1.apply$mcV$sp(PythonDStream.scala:101)
        at org.apache.spark.streaming.api.python.TransformFunction$$anonfun$writeObject$1.apply(PythonDStream.scala:100)
        at org.apache.spark.streaming.api.python.TransformFunction$$anonfun$writeObject$1.apply(PythonDStream.scala:100)
        at org.apache.spark.util.Utils$.tryOrIOException(Utils.scala:1303)
        ... 63 more</p>
  </blockquote>
  
  <p>Part of my code looks like this:</p>
  
  <pre><code>def sendPut(doc):
    res = es.index(index = "tweetrepository", doc_type= 'tweet', body = doc)
    return doc
  myJson = dataStream.map(decodeJson).map(addSentiment).map(sendPut)
  myJson.pprint()
  </code></pre>
  
  <p>Here is the decodeJson function:</p>
  
  <pre><code>def decodeJson(str):
    return json.loads(str)
  </code></pre>
  
  <p>Here is the addSentiment function:</p>
  
  <pre><code>def addSentiment(dic):
    dic['Sentiment'] = get_tweet_sentiment(dic['Text'])
    return dic
  </code></pre>
  
  <p>And here is the get_tweet_sentiment function:</p>
  
  <pre><code>def get_tweet_sentiment(tweet):
    analysis = TextBlob(tweet)
    if analysis.sentiment.polarity &gt; 0:
      return 'positive'
    elif analysis.sentiment.polarity == 0:
      return 'neutral'
    else:
      return 'negative'
  </code></pre>