<s> """<STR_LIT>""" <EOL> import argparse <EOL> import logging <EOL> from random import randrange <EOL> import google . cloud . dataflow as df <EOL> from google . cloud . dataflow . pvalue import AsIter <EOL> from google . cloud . dataflow . pvalue import AsList <EOL> from google . cloud . dataflow . pvalue import AsSingleton <EOL> def create_groups ( group_ids , corpus , word , ignore_corpus , ignore_word ) : <EOL> """<STR_LIT>""" <EOL> def attach_corpus_fn ( group , corpus , ignore ) : <EOL> selected = None <EOL> len_corpus = len ( corpus ) <EOL> while not selected : <EOL> c = corpus [ randrange ( <NUM_LIT:0> , len_corpus - <NUM_LIT:1> ) ] . values ( ) [ <NUM_LIT:0> ] <EOL> if c != ignore : <EOL> selected = c <EOL> yield ( group , selected ) <EOL> def attach_word_fn ( group , words , ignore ) : <EOL> selected = None <EOL> len_words = len ( words ) <EOL> while not selected : <EOL> c = words [ randrange ( <NUM_LIT:0> , len_words - <NUM_LIT:1> ) ] . values ( ) [ <NUM_LIT:0> ] <EOL> if c != ignore : <EOL> selected = c <EOL> yield group + ( selected , ) <EOL> return ( group_ids <EOL> | df . FlatMap ( <EOL> '<STR_LIT>' , <EOL> attach_corpus_fn , <EOL> AsList ( corpus ) , <EOL> AsSingleton ( ignore_corpus ) ) <EOL> | df . FlatMap ( <EOL> '<STR_LIT>' , <EOL> attach_word_fn , <EOL> AsIter ( word ) , <EOL> AsSingleton ( ignore_word ) ) ) <EOL> def run ( argv = None ) : <EOL> """<STR_LIT>""" <EOL> parser = argparse . ArgumentParser ( ) <EOL> parser . add_argument ( '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , default = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , default = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' ) <EOL> known_args , pipeline_args = parser . parse_known_args ( argv ) <EOL> p = df . Pipeline ( argv = pipeline_args ) <EOL> group_ids = [ ] <EOL> for i in xrange ( <NUM_LIT:0> , int ( known_args . num_groups ) ) : <EOL> group_ids . append ( '<STR_LIT:id>' + str ( i ) ) <EOL> query_corpus = '<STR_LIT>' <EOL> query_word = '<STR_LIT>' <EOL> ignore_corpus = known_args . ignore_corpus <EOL> ignore_word = known_args . ignore_word <EOL> pcoll_corpus = p | df . Read ( '<STR_LIT>' , <EOL> df . io . BigQuerySource ( query = query_corpus ) ) <EOL> pcoll_word = p | df . Read ( '<STR_LIT>' , <EOL> df . io . BigQuerySource ( query = query_word ) ) <EOL> pcoll_ignore_corpus = p | df . Create ( '<STR_LIT>' , [ ignore_corpus ] ) <EOL> pcoll_ignore_word = p | df . Create ( '<STR_LIT>' , [ ignore_word ] ) <EOL> pcoll_group_ids = p | df . Create ( '<STR_LIT>' , group_ids ) <EOL> pcoll_groups = create_groups ( pcoll_group_ids , pcoll_corpus , pcoll_word , <EOL> pcoll_ignore_corpus , pcoll_ignore_word ) <EOL> pcoll_groups | df . io . Write ( '<STR_LIT>' , <EOL> df . io . TextFileSink ( known_args . output ) ) <EOL> p . run ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> logging . getLogger ( ) . setLevel ( logging . INFO ) <EOL> run ( ) </s>
