<commit_before>import sys

from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from petrarch2 import EventCoder
from _functools import partial
#from CameoEventCoder import CameoEventCoder 

def map_articles(articleText):
    return articleText.encode('utf-8')
    

def code_articles(articleText, petrGlobals={}): 
    coder = EventCoder(petrGlobal = petrGlobals)
    events_map = coder.encode(articleText)
    return str(events_map)

def print_each(x):
    print x
    return x



if __name__ == "__main__":
    
  # create Spark context with Spark configuration
    conf = SparkConf().setAppName("Spark Petrarch2")
    sc = SparkContext(conf=conf)
    
    coder = EventCoder(petrGlobal={})
    
    bMap = sc.broadcast(coder.get_PETRGlobals())
    print(bMap.__class__)
    #sc.addPyFile("dist/petrarch2-1.0.0-py2.7.egg")
    #encoderBroadcast = sc.broadcast(CameoEventCoder())
    ssc = StreamingContext(sc, 120)
    kafkaStream = KafkaUtils.createStream(ssc=ssc,
                                        zkQuorum='dmlhdpc1',
                                        groupId='dmlhdpc1',
                                        topics={'petrarch':1})
  
 
    lines = kafkaStream.map(lambda x: x[1])
    events_rdd = lines.map(map_articles)
    events_rdd.pprint(1)
    events_rdd = events_rdd.map(partial(code_articles, bMap.value))
    events_rdd.pprint(1)
    
    events_rdd.saveAsTextFiles("hdfs://dmlhdpc10:9000/Events_SPEC", "OUT")
<commit_msg>FInal Code For Testing<commit_after>import sys

from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from petrarch2 import EventCoder
from _functools import partial
#from CameoEventCoder import CameoEventCoder 

def map_articles(articleText):
    return articleText.encode('utf-8')
    

def code_articles(articleText, petrGlobals={}): 
    coder = EventCoder(petrGlobal = petrGlobals)
    events_map = coder.encode(articleText)
    return str(events_map)

def print_each(x):
    print x
    return x



if __name__ == "__main__":
    
  # create Spark context with Spark configuration
    conf = SparkConf().setAppName("Spark Petrarch2")
    sc = SparkContext(conf=conf)
    
    coder = EventCoder(petrGlobal={})
    
    bMap = sc.broadcast(coder.get_PETRGlobals())
    print(bMap.__class__)
    #sc.addPyFile("dist/petrarch2-1.0.0-py2.7.egg")
    #encoderBroadcast = sc.broadcast(CameoEventCoder())
    ssc = StreamingContext(sc, 120)
    kafkaStream = KafkaUtils.createStream(ssc=ssc,
                                        zkQuorum='dmlhdpc1',
                                        groupId='dmlhdpc1',
                                        topics={'petrarch':1})
  
 
    lines = kafkaStream.map(lambda x: x[1])
    events_rdd = lines.map(map_articles)
    events_rdd.pprint(1)
    events_rdd = events_rdd.map(partial(code_articles, petrGlobals = bMap.value))
    events_rdd.pprint(1)
    
    events_rdd.saveAsTextFiles("hdfs://dmlhdpc10:9000/Events_SPEC", "OUT")

{'fraction_non_alphanumeric': 0.09906421426266537, 'fraction_numerical': 0.015488867376573089, 'mean_word_length': 3.0469973890339426, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4580744', 'n_tokens_mistral': 1087, 'n_tokens_neox': 1041, 'n_words': 184}