<s> from pyspark import SparkContext , SparkConf <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> conf = SparkConf ( ) . setAppName ( "<STR_LIT>" ) <EOL> sc = SparkContext ( conf = conf ) <EOL> es_read_conf = { <EOL> "<STR_LIT>" : "<STR_LIT:localhost>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> es_write_conf = { <EOL> "<STR_LIT>" : "<STR_LIT:localhost>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> es_rdd = sc . newAPIHadoopRDD ( <EOL> inputFormatClass = "<STR_LIT>" , <EOL> keyClass = "<STR_LIT>" , <EOL> valueClass = "<STR_LIT>" , <EOL> conf = es_read_conf ) <EOL> doc = es_rdd . first ( ) [ <NUM_LIT:1> ] <EOL> for field in doc : <EOL> value_counts = es_rdd . map ( lambda item : item [ <NUM_LIT:1> ] [ field ] ) <EOL> value_counts = value_counts . map ( lambda word : ( word , <NUM_LIT:1> ) ) <EOL> value_counts = value_counts . reduceByKey ( lambda a , b : a + b ) <EOL> value_counts = value_counts . filter ( lambda item : item [ <NUM_LIT:1> ] > <NUM_LIT:1> ) <EOL> value_counts = value_counts . map ( lambda item : ( '<STR_LIT:key>' , { <EOL> '<STR_LIT>' : field , <EOL> '<STR_LIT>' : item [ <NUM_LIT:0> ] , <EOL> '<STR_LIT:count>' : item [ <NUM_LIT:1> ] <EOL> } ) ) <EOL> value_counts . saveAsNewAPIHadoopFile ( <EOL> path = '<STR_LIT:->' , <EOL> outputFormatClass = "<STR_LIT>" , <EOL> keyClass = "<STR_LIT>" , <EOL> valueClass = "<STR_LIT>" , <EOL> conf = es_write_conf ) </s>
