from textblob import TextBlob as tb<br>
from textblob_aptagger import PerceptronTagger<br>
import numpy as np<br>
import nltk.data<br>
import Constants<br>
from pyspark import SparkContext,SparkConf<br>
import nltk<br>

TOKENIZER = nltk.data.load('tokenizers/punkt/english.pickle')

def word_tokenize(x):<br>
   &nbsp; return nltk.word_tokenize(x)

def pos_tag (s):<br>
  &nbsp;global TAGGER<br>
  &nbsp;return TAGGER.tag(s)<br>

def wrap_words (pair):<br>
  &nbsp;''' associable each word with index '''<br>
  &nbsp;index = pair[0]<br>
  &nbsp;result = []<br>
  &nbsp;for word, tag in pair[1]:<br>
    &nbsp;&nbsp;word = word.lower()<br>
    &nbsp;&nbsp;result.append({ "index": index, "word": word, "tag": tag})<br>
    &nbsp;&nbsp;index += 1<br>
  &nbsp;return result<br>

if __name__ == '__main__':

  &nbsp;conf = SparkConf().setMaster(Constants.MASTER_URL).setAppName(Constants.APP_NAME)<br>
  &nbsp;sc = SparkContext(conf=conf)<br>
  &nbsp;data = sc.textFile(Constants.FILE_PATH)<br>

  &nbsp;sent = data.flatMap(word_tokenize).map(pos_tag).map(lambda x: x[0]).glom()<br>
  &nbsp;num_partition = sent.getNumPartitions()<br>
  &nbsp;<b>base = list(np.cumsum(np.array(sent.map(len).collect())))<br></b>
  &nbsp;base.insert(0, 0)<br>
  &nbsp;base.pop()<br>
  &nbsp;RDD = sc.parallelize(base,num_partition)<br>
  &nbsp;tagged_doc = RDD.zip(sent).map(wrap_words).cache()<br>