Q:

How do I map the stream filter function of tweepy with the thread pool? 

Using the library:
from multiprocessing.dummy import Pool as ThreadPool

I am streaming tweets and then processing them as follows:
class StreamListener(tweepy. StreamListener):
httpsCheck = 'https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+'
httpCheck = 'http?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+'

def on_status(self, status):
    if status.retweeted:
        return
    tweetText = status.text.encode('utf8')
    created_at = status.created_at
    id = status.id
    if (re.findall(self.httpCheck, tweetText) or re.findall(self.httpsCheck, tweetText)):
        return
    if (re.search('[a-zA-Z]', tweetText)):
        response = natural_language_understanding.analyze(
            text=tweetText,
            features=Features(
                entities=EntitiesOptions(
                    emotion=True,
                    sentiment=True,
                    limit=2),
                keywords=KeywordsOptions(
                    emotion=True,
                    sentiment=True,
                    limit=2)),
            language='en'
            )
        response["tweet"] = tweetText
        response["id"] = id
        response["created_at"] = created_at
        with open('#LFCROMA-SF2.csv', 'a') as csv_file:
            writer = csv.writer(csv_file)
            for key, value in response.items():
                writer.writerow([key, value])

def on_error(self, status_code):
    if status_code == 420:
        return False

Now, I want to use multi-threading in order to increase my speed using pool
auth = tweepy. OAuthHandler(settingsTwitter. TWITTER_APP_KEY, settingsTwitter. TWITTER_APP_SECRET)
auth.set_access_token(settingsTwitter. TWITTER_KEY, settingsTwitter. TWITTER_SECRET)
api = tweepy. API(auth)
stream_listener = StreamListener()
stream = tweepy. Stream(auth=api.auth, listener=stream_listener)
pool = ThreadPool(4)
pool.map(stream.filter, languages=["en"], track=["#LFC"])

By doing so, I am getting an error: 

Traceback (most recent call last):   File "atomic.py", line 66, in
  
      pool.map(stream.filter, languages=["en"], track=["#LFC"]) TypeError: map() got an unexpected keyword argument 'languages'

I am unable to figure out a way to call this function in pool.map()
The purpose is to stream tweets, apply some heavy processing and then save the results.  There is a bottleneck as streaming >> processing in terms of speed.  Which is why I want to stream tweets in a multithreaded approach. 

A:

The best way to have this sorted out is:

Stream tweets seperately and dont use threading on them. 
Write a function which would read the tweets and then send it for processing(Watson NLU processing in this case)
The reading of the tweets from the file in tail -f manner in a parallel manner(multi threading) will ensure that there is no processing bottleneck. 

To understand this with an eg:
# -*- coding: utf-8 -*-
import re
import csv
import json
import settingsWatson
from watson_developer_cloud import NaturalLanguageUnderstandingV1
from watson_developer_cloud.natural_language_understanding_v1 import Features, EntitiesOptions, KeywordsOptions, CategoriesOptions
import os
import time
import settingsTwitter
import tweepy
import datetime
from multiprocessing.dummy import Pool as ThreadPool

natural_language_understanding = NaturalLanguageUnderstandingV1(
  username=settingsWatson.username,
  password=settingsWatson.password,
  version='2018-03-16')

class StreamListener(tweepy. StreamListener):
  tweet = {}
  httpsCheck = 'https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+'
  httpCheck = 'http?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+'
  idSelf = 0
  fieldNames = ['tweet', 'sequence', 'created_at', 'id']

  def on_status(self, status):
    if status.retweeted:
      return
    tweetText = status.text.encode('utf8')
    created_at = status.created_at
    id = status.id
    if (re.findall(self.httpCheck, tweetText) or re.findall(self.httpsCheck, tweetText)):
      return
    if (re.search('[a-zA-Z]', tweetText)):
      self.idSelf += 1
      self.tweet["tweet"] = tweetText
      self.tweet["id"] = id
      self.tweet["sequence"] = self.idSelf
      self.tweet["created_at"] = created_at
      with open('#ELCLASICO-2018-05-07.csv', 'a') as csv_file:
        #json.dump(self.tweet, json_file, sort_keys=True, indent=4, default = str)
        writer = csv. DictWriter(csv_file, self.tweet.keys())
        #for key, value in self.tweet.items():
        #a = [self.tweet]
        #print a[0]['tweet']
        writer.writerow(self.tweet)

  def on_error(self, status_code):
    if status_code == 420:
      return False

auth = tweepy. OAuthHandler(settingsTwitter. TWITTER_APP_KEY, settingsTwitter. TWITTER_APP_SECRET)
auth.set_access_token(settingsTwitter. TWITTER_KEY, settingsTwitter. TWITTER_SECRET)
api = tweepy. API(auth)
print "Twitter API Authentication is successful!" 
stream_listener = StreamListener()
stream = tweepy. Stream(auth=api.auth, listener=stream_listener)
print "Streaming begins!" 

def startStream():
  while True:
    try:
      stream = tweepy. Stream(auth=api.auth, listener=stream_listener)
      stream.filter(languages=["en"], track=["#ElClasico"])
    except:
      continue

startStream()

The above code is used to stream tweets with certain conditions and then save them in a CSV file.  
However, this was never an issue.  The issue was in my "atomic" approach.  I should have avoided the bottleneck not by bringing the whole procedure(stream, process, save) in one but by distributed computing approach(stream is one task, process is another).  Note, that this method fails as well(since the processing can never match the speed of the stream).  In order to overcome this, I applied multithreading in a distributed fashion:

Stream tweets
Save tweets
[this (below) is the point from where multithreading approach starts]
Read the tweets in tail -f manner
Send them to Watson for processing
Perform Data File Handling and stuff

This is the code for processing stuff:
# -*- coding: utf-8 -*-
import re
import csv
import json
import settingsWatson
from watson_developer_cloud import NaturalLanguageUnderstandingV1
from watson_developer_cloud.natural_language_understanding_v1 import Features, EntitiesOptions, KeywordsOptions, CategoriesOptions
import os
import time
import settingsTwitter
import tweepy
import datetime
from multiprocessing.dummy import Pool as ThreadPool

natural_language_understanding = NaturalLanguageUnderstandingV1(
    username=settingsWatson.username,
    password=settingsWatson.password,
    version='2018-03-16')

class FileTailer(object):

    def __init__(self, file, delay=1):
        self.file = file
        self.delay = delay

    def __iter__(self):
        while True:
            where = self.file.tell()
            line = self.file.readline()
            if line and line.endswith('\n'): # only emit full lines
                yield line
                #response = self.naturalLanguageProcessing(line)
            else:
                print "Waiting for new line"
                # for a partial line, pause and back up
                time.sleep(self.delay)       # ...not actually a recommended approach. 
                self.file.seek(where)

class watson:

  entityDict = {"Messi":["Lionel Messi", "Leo", "Messi"], "Ronaldo":["Cristiano Ronaldo", "Cristiano", "Ronaldo"], "Iniesta":["Andres Iniesta", "Iniesta"], "Barcelona":["Barca", "Barcelona", "FC Barcelona", "#FCBarcelona"], "Real Madrid":["Real Madrid", "Madrid", "#RMA", "#RealMadrid"]}
  date = "2018-05-06"

  def createFiles(self):
    for entity in self.entityDict:
      fileName = str(entity) + "-" + str(self.date) + ".csv"
      print fileName
      with open(fileName, 'wb') as myFile:
        wr = csv.writer(myFile, quoting=csv. QUOTE_ALL)
    print "This is the entity defined list:"
    print self.entityDict
    for i in self.entityDict:
      for j in self.entityDict[i]:
        print j

  def naturalLanguageProcessing(self,tweetText):
    print "NLP is called with this text: " + tweetText
    try:
      response = natural_language_understanding.analyze(
        text=tweetText,
        features=Features(
          entities=EntitiesOptions(
            emotion=True,
            sentiment=True,
            limit=2),
          keywords=KeywordsOptions(
            emotion=True,
            sentiment=True,
            limit=2)),
        language='en'
        )
      response["tweet"] = tweetText
      self.saveResults(response)
    except:
      print "Error occured.  Sleeping for one"
      time.sleep(1)
      return None

  def saveResults(self, response):
    print "Saving the results"
    entitiesTweet = response["entities"]
    print "Printing entitiesTweet"
    print entitiesTweet
    for entity in entitiesTweet:
      try:
        for i in self.entityDict:
          for j in self.entityDict[i]:
            if(j == entity["text"]):
              fileName = str(self.entityDict[i]) + "-" + str(self.date) + ".csv"
              with open(fileName, 'a') as myFile:
                wr = csv.writer(myFile, quoting=csv. QUOTE_ALL)
                entity["tweet"] = response["tweet"]
                wr.writerow([entity])
      except Exception as e:
        print (e)

          #entityToBeInserted = entity
          #entityToBeInserted["tweet"] = response["tweet"]
          #fileName = str(entityItem) + "-" + str(date) + ".csv"
          #with open(fileName, 'a') as myFile:
            #wr = csv.writer(myfile, quoting=csv. QUOTE_ALL)
            #wr.writerow([entityToBeInserted])
    with open('#ELCLASICO-2018-05-07-Watson.csv', 'a') as csv_file:
      writer = csv.writer(csv_file)
      writer.writerow([response])

  #def saveEntityResults(self, entities, date):

def initiator(farzi):
  csv_reader = csv.reader(FileTailer(open('#ELCLASICO-2018-05-07.csv')))
  ob = watson()
  for row in csv_reader:
    tweet = row[1]
    ob.naturalLanguageProcessing(tweet)

pool = ThreadPool(4) 
farzi = "farzi"
pool.map(initiator, farzi)

