<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Install ES-Hadoop connector
Download ES-Hadoop for Spark 2.0+ We are using 5.2.0.
Add  either elasticsearch-hadoop-5.2.0.jar or  elasticsearch-spark-20_2.10-5.2.0.jar  to Spark  classpath
Step2:  Goal
Step3:  Configure  ES  parameters
Mandatory
Step4:  ES returns key-value RDD where key is ID of the document, and value is content of _source field 
Step5:  Every JSON document  represents a connection between a  source IP and  destination IP.
Step6:  List all the  unique ip addresses
Step7:  Create  two columns
rdd_incoming
Step8:  Join the two RDD's to create a dataframe like  structure
Step9:  Replace missing values with zeroes
Step10:  Gather statistics  on our dataset
Step11:  Identify  IPs that  have  unusual  number of  outgoing or incoming connections
Step12:  Clean up
<ASSISTANT_TASK:>
Python Code:
import os
import sys
sys.path.append(os.environ["SPARK_HOME"] + "/python/lib/py4j-0.10.4-src.zip")
sys.path.append(os.environ["SPARK_HOME"] + "/python/lib/pyspark.zip")
from pyspark import SparkConf, SparkContext
sconf = SparkConf()
sconf.setAppName("ES-Spark Integration")
sconf.setMaster("spark://snehasish-barmans-macbook.local:7077")
sc = SparkContext.getOrCreate(conf = sconf)
print sc
print sc.version
Explanation: Install ES-Hadoop connector
Download ES-Hadoop for Spark 2.0+ We are using 5.2.0.
Add  either elasticsearch-hadoop-5.2.0.jar or  elasticsearch-spark-20_2.10-5.2.0.jar  to Spark  classpath:
Edit  the  conf/spark-env.sh file  to add them  to classpath. For instance:
SPARK_CLASSPATH={SPARK_HOME}/jars/elasticsearch-hadoop/elasticsearch-hadoop-5.2.0.jar
Using PySpark with ES-Hadoop  
We are using Spark 2.1.0. If using Spark 1.6 or below, download the appropriate hadoop connector (2.4).
End of explanation
def getESQuery():
    query = {
        "query": {
                "filtered": {
                     "query": {
                         "match_all": {}
                      },
                      "filter": {
                          "bool": {
                             "must": [
                                 {
                                    "range": {
                                        "@timestamp" : {
                                           "gte" : "now-1h",
                                           "lt" : "now"
                                        }
                                    }
                                 }
                              ],
                              "should": [
                                 { "term": { "src_addr_space": "private" } },
                                 { "term": { "dst_addr_space": "private" } }
                              ]
                          }
                    }
                }
           },
        "_source": [ "netflow.ipv4_src_addr", "netflow.ipv4_dst_addr" ]
        }
    return query
Explanation: Goal:  Identify  3-sigma  deviation connections
Construct  ES Query
Query DSL (Here, we select connections between machines that operate within a restricted environment)
End of explanation
ES_HOST = "192.168.19.156"
ES_PORT = "9200"
ES_INDEX = "seceon_flows"
ES_QUERY = getESQuery()
es_config = {"es.nodes": ES_HOST,
             "es.port": ES_PORT,
             "es.resource": ES_INDEX,
             "es.query": ES_QUERY,
             "es.nodes.discovery": "false",
             "es.nodes.wan.only": "true"
            }
esrdd = sc.newAPIHadoopRDD(inputFormatClass = "org.elasticsearch.hadoop.mr.EsInputFormat",
                           keyClass = "org.apache.hadoop.io.NullWritable",
                           valueClass = "org.elasticsearch.hadoop.mr.LinkedMapWritable",
                           conf = es_config)
Explanation: Configure  ES  parameters
Mandatory: ES-host, ES-port,  ES-index, ES-query
If  operating in a cloud/restricted environment and want to disable node discovery,  set es.nodes.discovery and  es.nodes.wan.only as shown below.
More Info here
End of explanation
print esrdd.getNumPartitions()
esrdd = esrdd.coalesce(4)
print esrdd.getNumPartitions()
esrdd.persist()
print esrdd.is_cached
print esrdd.getStorageLevel()
rdd1 = esrdd.sample(False, 0.01).mapValues( lambda x: {"src": x["netflow"]["ipv4_src_addr"], 
                                                                    "dst": x["netflow"]["ipv4_dst_addr"]} )
rdd1.take(4)
Explanation: ES returns key-value RDD where key is ID of the document, and value is content of _source field :  [("_id", {}), ...]
End of explanation
rdd1.countApprox(1000, 0.1)
Explanation: Every JSON document  represents a connection between a  source IP and  destination IP.
End of explanation
print rdd1.flatMap( lambda x: [ x[1]["src"], x[1]["dst"] ] ).distinct().count()
print rdd1.flatMap( lambda x: [ x[1]["src"], x[1]["dst"] ] ).distinct().collect()
Explanation: List all the  unique ip addresses
End of explanation
rdd_outgoing = rdd1.map( lambda x: (x[1]["src"], 1) ).reduceByKey( lambda acc, x: acc + x )
rdd_outgoing.collect()
rdd_incoming = rdd1.map( lambda x: (x[1]["dst"], 1) ).reduceByKey( lambda acc, x: acc + x )
rdd_incoming.collect()
Explanation: Create  two columns
rdd_incoming:  RDD of counts of incoming connections (received) for every IP
rdd_outgoing:  RDD of counts of outgoing connections (sent) for every IP
End of explanation
rdd_joined = rdd_outgoing.fullOuterJoin(rdd_incoming)
rdd_joined.collect()
Explanation: Join the two RDD's to create a dataframe like  structure
End of explanation
rdd_joined = rdd_joined.mapValues(lambda x: tuple([y if y is not None else 0 for y in x]))
rdd_joined.collect()
Explanation: Replace missing values with zeroes
End of explanation
src_stats = rdd_joined.map(lambda x: x[1][0]).stats() # outgoing(src)
print src_stats
dst_stats = rdd_joined.map(lambda x: x[1][1]).stats() # incoming(dst)
print dst_stats
src_stats_bc = sc.broadcast(src_stats)
dst_stats_bc = sc.broadcast(dst_stats)
src_stats_bc.value
Explanation: Gather statistics  on our dataset
End of explanation
rdd_joined.filter(lambda x: x[1][0] > 0).filter( lambda x: (x[1][0] - src_stats_bc.value.mean()) > (3*src_stats_bc.value.stdev()) ).collect()
rdd_joined.filter(lambda x: x[1][1] > 0).filter( lambda x: (x[1][1] - dst_stats_bc.value.mean()) > (3*dst_stats_bc.value.stdev()) ).collect()
Explanation: Identify  IPs that  have  unusual  number of  outgoing or incoming connections
End of explanation
src_stats_bc.unpersist()
dst_stats_bc.unpersist()
esrdd.unpersist()
print esrdd.is_cached
sc.stop()
Explanation: Clean up
End of explanation
<END_TASK>