<s> import sys <EOL> import json <EOL> from pyspark import SparkContext , SparkConf <EOL> from elasticsearch import Elasticsearch <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> task_id = sys . argv [ <NUM_LIT:1> ] <EOL> print ( task_id ) <EOL> ES_HOST = { <EOL> "<STR_LIT:host>" : "<STR_LIT:localhost>" , <EOL> "<STR_LIT:port>" : <NUM_LIT> <EOL> } <EOL> es = Elasticsearch ( hosts = [ ES_HOST ] ) <EOL> es . update ( index = '<STR_LIT>' , doc_type = '<STR_LIT>' , id = task_id , body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT:status>' : '<STR_LIT>' <EOL> } <EOL> } ) <EOL> result_indices = len ( es . indices . get_aliases ( index = "<STR_LIT>" ) ) <EOL> output_resource = "<STR_LIT>" % ( result_indices + <NUM_LIT:1> ) <EOL> conf = SparkConf ( ) . setAppName ( "<STR_LIT>" ) <EOL> sc = SparkContext ( conf = conf ) <EOL> es_rdd = sc . newAPIHadoopRDD ( <EOL> inputFormatClass = "<STR_LIT>" , <EOL> keyClass = "<STR_LIT>" , <EOL> valueClass = "<STR_LIT>" , <EOL> conf = { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> doc = es_rdd . first ( ) [ <NUM_LIT:1> ] <EOL> num_fields = len ( doc ) <EOL> for idx , field in enumerate ( doc ) : <EOL> es . update ( index = '<STR_LIT>' , doc_type = '<STR_LIT>' , id = task_id , body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : ( idx + <NUM_LIT:1> ) * <NUM_LIT> / num_fields , <EOL> '<STR_LIT:status>' : '<STR_LIT>' <EOL> } <EOL> } ) <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 = { "<STR_LIT>" : output_resource } ) <EOL> es . update ( index = '<STR_LIT>' , doc_type = '<STR_LIT>' , id = task_id , body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT:100> , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:result>' : output_resource <EOL> } <EOL> } ) </s>
