[PySpark][Streaming] Python API for Spark Streaming

# Python API for Spark Streaming

### Implementation Concepts
* Not to modify existing pyspark code aggressively.
* Reuse pysaprk code.
* Use a py4j callback server for interaction between Python and Java/Scala.

### Functions
Following functions are supported.

**Input streams:**
* textFileStream
* socketTextStream

**DStream Operations:**
* context()
* combineByKey()
* reduceByKey()
* filter()
* flatMap()
* map()
* print()
* reduce()
* count()
* groupByKey()
* saveAsTextFiles() writing test case
* saveAsPickledFiles() writing test case
* mapValues()
* flatMapValues()
* glom()
* mapPartitions()
* countByValue()
* foreachRDD()
<hr>
* checkpoint() no test case yet
* cache() no test case yet
* persist() no test case yet


### Examples:
1. Network wordcount
```
     ./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py <host> <port>
     
     # Terminal 1
     $ nc -lk 9999
     
     # Terminal 2
     $ ./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999
```
2. Wordcount
```
     $ ./bin/spark-submit examples/src/main/python/streaming/wordcount.py <directory_path>
```

### Known issues and discussions

1  **Name of PythonPairwiseDStream**

This class has the same functionality PairwiseRDD In Python API for Spark. I think that the convention of api should have prefix, like Java, Python. Perhaps, PairwiseRDD does not need “Python” prefix because PairwiseRDD does not exist in Scala/Java. 

Which name is appropriate?
PythonPairwiseDStream or PairwiseDStream

2  **pyspark.streaming.dstream.print_ pyprint**

I implement the function called print_ and pyprint. print_ calls DStream.print in Scala and shows byte array because PythonRDD keeps data as ArrayByte. pyprint is the same function to DStream.print in Scala.

Do we need print_ ?

3 **Testcase for checkpoint, persist, cache**

While RDD has  getCheckpointFile, DStream has checkpointData.currentCheckpointFiles. But I don't figure out how to access from python.

4  **Callback server**

Especially when error is happened, call backer server still alive after terminate the process. It occupies port 25334(default port of py4j callback server)

How to kill callback server manually (in Mac)
```
kill -9 $(sudo lsof -i TCP | grep LISTEN | grep 25334 | awk -F' ' '{print $2}’)
```

Perhaps, this error should be handled by application layer like this:
```
try
    ssc = StreamingContext()
    ...
    ssc.start()
    ssc.awaitTermination()
finally
    ssc.stop()
```

5  **py4j bugs?**

This error sometimes could be happened during test case. This is not test case failure. Please retry the test.
```
py4j.Py4JException: An exception was raised by the Python Proxy. Return Message: null
     at py4j.Protocol.getReturnValue(Protocol.java:417)
     at py4j.reflection.PythonProxyHandler.invoke(PythonProxyHandler.java:113)
     at com.sun.proxy.$Proxy6.call(Unknown Source)
     at org.apache.spark.streaming.api.python.PythonForeachDStream$$anonfun$$init$$1.apply(PythonDStream.scala:108)
     at org.apache.spark.streaming.api.python.PythonForeachDStream$$anonfun$$init$$1.apply(PythonDStream.scala:107)
     at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply$mcV$sp(ForEachDStream.scala:41)
     at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40)
     at org.apache.spark.streaming.dstream.ForEachDStream$$anonfun$1.apply(ForEachDStream.scala:40)
     at scala.util.Try$.apply(Try.scala:161)
     at org.apache.spark.streaming.scheduler.Job.run(Job.scala:32)
     at org.apache.spark.streaming.scheduler.JobScheduler$JobHandler.run(JobScheduler.scala:172)
     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
     at java.lang.Thread.run(Thread.java:745)
```

6 **How to implement transform?**

py4j callback function does not return JavaObject as far as I know. Every batch time, JVM pass JavaRDD(reference) to callback server and does computation based on user-defined functions. It should return JavaRDD(reference) to JVM. In addition, deserialize should be changed dynamically. Currently, it is defined by beginning.

We need to figure out how to implement transform operations.
* transform
* transoformWith

Following operations have dependency to transform in current implementation.
* repartions()
* join() 
* leftOuterJoin() 
* rightOuterJoin() 
* cogroup()
* union() 


7 **How to get number of partition? Currently hardcoded.**

We have not figure out how to get number of partitions through DStream from python.

```
    def getNumPartitions(self):
        """
        Return the number of partitions in RDD
        """
        # TODO: remove hard coding. RDD has NumPartitions. How do we get the number of partition
        # through DStream?
        return 2
```