<s> from itertools import tee , izip <EOL> from nltk import stem <EOL> from text . blob import TextBlob as tb <EOL> from textblob_aptagger import PerceptronTagger <EOL> import nltk . data <EOL> import numpy as np <EOL> import sys <EOL> TOKENIZER = nltk . data . load ( '<STR_LIT>' ) <EOL> TAGGER = PerceptronTagger ( ) <EOL> STEMMER = stem . porter . PorterStemmer ( ) <EOL> def pos_tag ( s ) : <EOL> """<STR_LIT>""" <EOL> global TAGGER <EOL> return TAGGER . tag ( s ) <EOL> def wrap_words ( pair ) : <EOL> """<STR_LIT>""" <EOL> global STEMMER <EOL> index = pair [ <NUM_LIT:0> ] <EOL> result = [ ] <EOL> for word , tag in pair [ <NUM_LIT:1> ] : <EOL> word = word . lower ( ) <EOL> stem = STEMMER . stem ( word ) <EOL> if stem == "<STR_LIT>" : <EOL> stem = word <EOL> keep = tag in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , ) <EOL> result . append ( { "<STR_LIT:id>" : <NUM_LIT:0> , "<STR_LIT:index>" : index , "<STR_LIT>" : stem , "<STR_LIT>" : word , "<STR_LIT>" : tag , "<STR_LIT>" : keep } ) <EOL> index += <NUM_LIT:1> <EOL> return result <EOL> TEXT = """<STR_LIT>""" <EOL> from pyspark import SparkContext <EOL> sc = SparkContext ( appName = "<STR_LIT>" , master = "<STR_LIT>" ) <EOL> sent = sc . parallelize ( TOKENIZER . tokenize ( TEXT ) ) . map ( pos_tag ) . cache ( ) <EOL> sent . collect ( ) <EOL> base = list ( np . cumsum ( np . array ( sent . map ( len ) . collect ( ) ) ) ) <EOL> base . insert ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> base . pop ( ) <EOL> sent_length = sc . parallelize ( base ) <EOL> tagged_doc = sent_length . zip ( sent ) . map ( wrap_words ) <EOL> from pyspark . sql import SQLContext , Row <EOL> sqlCtx = SQLContext ( sc ) <EOL> def enum_words ( s ) : <EOL> for word in s : <EOL> yield word <EOL> words = tagged_doc . flatMap ( enum_words ) <EOL> pair_words = words . keyBy ( lambda w : w [ "<STR_LIT>" ] ) <EOL> uniq_words = words . map ( lambda w : w [ "<STR_LIT>" ] ) . distinct ( ) . zipWithUniqueId ( ) <EOL> uniq = sc . broadcast ( dict ( uniq_words . collect ( ) ) ) <EOL> def id_words ( pair ) : <EOL> ( key , val ) = pair <EOL> word = val [ <NUM_LIT:0> ] <EOL> id = val [ <NUM_LIT:1> ] <EOL> word [ "<STR_LIT:id>" ] = id <EOL> return word <EOL> id_doc = pair_words . join ( uniq_words ) . map ( id_words ) <EOL> id_words = id_doc . map ( lambda w : ( w [ "<STR_LIT:id>" ] , w [ "<STR_LIT:index>" ] , w [ "<STR_LIT>" ] , w [ "<STR_LIT>" ] , w [ "<STR_LIT>" ] ) ) <EOL> wordSchema = id_words . map ( lambda p : Row ( id = long ( p [ <NUM_LIT:0> ] ) , index = int ( p [ <NUM_LIT:1> ] ) , word = p [ <NUM_LIT:2> ] , stem = p [ <NUM_LIT:3> ] , tag = p [ <NUM_LIT:4> ] ) ) <EOL> wordTable = sqlCtx . inferSchema ( wordSchema ) <EOL> wordTable . registerTempTable ( "<STR_LIT>" ) <EOL> wordTable . saveAsParquetFile ( "<STR_LIT>" ) <EOL> def sliding_window ( iterable , size ) : <EOL> """<STR_LIT>""" <EOL> iters = tee ( iterable , size ) <EOL> for i in xrange ( <NUM_LIT:1> , size ) : <EOL> for each in iters [ i : ] : <EOL> next ( each , None ) <EOL> return list ( izip ( * iters ) ) <EOL> def keep_pair ( pair ) : <EOL> """<STR_LIT>""" <EOL> return pair [ <NUM_LIT:0> ] [ "<STR_LIT>" ] and pair [ <NUM_LIT:1> ] [ "<STR_LIT>" ] and ( pair [ <NUM_LIT:0> ] [ "<STR_LIT>" ] != pair [ <NUM_LIT:1> ] [ "<STR_LIT>" ] ) <EOL> def link_words ( seq ) : <EOL> """<STR_LIT>""" <EOL> return [ ( seq [ <NUM_LIT:0> ] , word ) for word in seq [ <NUM_LIT:1> : ] ] <EOL> tiled = tagged_doc . flatMap ( lambda s : sliding_window ( s , <NUM_LIT:3> ) ) . flatMap ( link_words ) . filter ( keep_pair ) <EOL> t0 = tiled . map ( lambda l : ( uniq . value [ l [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ] , uniq . value [ l [ <NUM_LIT:1> ] [ "<STR_LIT>" ] ] , ) ) <EOL> t1 = tiled . map ( lambda l : ( uniq . value [ l [ <NUM_LIT:1> ] [ "<STR_LIT>" ] ] , uniq . value [ l [ <NUM_LIT:0> ] [ "<STR_LIT>" ] ] , ) ) <EOL> neighbors = t0 . union ( t1 ) <EOL> edgeSchema = neighbors . map ( lambda p : Row ( n0 = long ( p [ <NUM_LIT:0> ] ) , n1 = long ( p [ <NUM_LIT:1> ] ) ) ) <EOL> edgeTable = sqlCtx . inferSchema ( edgeSchema ) <EOL> edgeTable . registerTempTable ( "<STR_LIT>" ) <EOL> edgeTable . saveAsParquetFile ( "<STR_LIT>" ) </s>
