[PySpark][Streaming][SPARK-2377] Python API for Spark Streaming

I will post most of the  same comment below when I PR to master in apache/spark. If something is wrong, please let me know.


# Python API for Spark Streaming

### Implementation Concepts
* Not to modify existing pyspark code aggressively.
* Reuse pysaprk code.
* Use a py4j callback server for an interaction between Python and Java/Scala.

### Functions
Following functions are supported.

**Input streams:**
* textFileStream
* socketTextStream

**DStream operations:**
* context()
* combineByKey()
* reduceByKey()
* filter()
* flatMap()
* map()
* pyprint(): the same function to 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
If we solve following topics in this thread, I do not comment these topic in PR to master.

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 is a reserved name for python:) 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 passes JavaRDD(reference) to callback server and does computation based on user-defined functions. Callback server 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 figured 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
```

8 **How to wrap Time class in Scala with Python**
Time class would be time.py in python if I follow the naming convention for python script in pyspark. time.py conflict to python default time package.

What is the proper name?
Currently there is no use case for Time from python. I can remove from this PR for now.

9 **When do we update documentation?**

pyspark.streaming is still (pre) beta version. If we update the documentation, more people know this python API. However, we have not figured out how to integrate Kafka, Twitter and so on. 

If TD wants me update documentation, I will do that.  

10 **How many JIRA should I create?**

* Implement transform
* Support Twitter, Kafak etc..


Take a look at https://github.com/apache/spark/pull/1628#issuecomment-50663192 for a discussion of a similar issue.  It's possible to name this `time.py` without any conflicts; look at libraries like `numpy`, where you have `numpy.random`, to see that this is possible.