<s> import time <EOL> import logging <EOL> from dpark . dstream import StreamingContext <EOL> from dpark import optParser <EOL> def load_context ( checkpointDir = None ) : <EOL> if checkpointDir : <EOL> try : <EOL> ssc , start_time = StreamingContext . load ( checkpointDir ) <EOL> return ssc , start_time <EOL> except : <EOL> logging . warning ( '<STR_LIT>' , checkpointDir ) <EOL> ssc = StreamingContext ( <NUM_LIT:1> ) <EOL> return ssc , None <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> options , args = optParser . parse_args ( ) <EOL> host , port = args <EOL> port = int ( port ) <EOL> ssc , start_time = load_context ( options . checkpoint_dir ) <EOL> if start_time is None : <EOL> ssc . checkpoint ( options . checkpoint_dir , <NUM_LIT:20> ) <EOL> st = ssc . networkTextStream ( host , port ) . flatMap ( lambda x : x . split ( ) ) . map ( lambda x : ( x , <NUM_LIT:1> ) ) . updateStateByKey ( lambda vs , state : sum ( vs ) + ( state or <NUM_LIT:0> ) ) . show ( ) <EOL> ssc . start ( start_time ) <EOL> ssc . awaitTermination ( ) </s>
