How to use Spark Streaming to read a stream and find the IP over a time Window?

Question

I am new to Apache Spark and I would like to write some code in Python using PySpark to read a stream and find the IP addresses.

I have a Java class to generate some fake ip addresses in order to process them afterwards. This class will be listed here:

import java.io.DataOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Random;

public class SocketNetworkTrafficSimulator {
    public static void main(String[] args) throws Exception {
        Random rn = new Random();
        ServerSocket welcomeSocket = new ServerSocket(9999);
        int[] possiblePortTypes = new int[]{21, 22, 80, 8080, 463};
        int numberOfRandomIps=100;
        String[] randomIps = new String[numberOfRandomIps];
        for (int i=0;i<numberOfRandomIps;i++)
            randomIps[i] = (rn.nextInt(250)+1) +"." +
                                (rn.nextInt(250)+1) +"." +
                                (rn.nextInt(250)+1) +"." +
                                (rn.nextInt(250)+1);
        System.err.println("Server started");
        while (true) {
            try {
                Socket connectionSocket = welcomeSocket.accept();
                System.err.println("Server accepted connection");
                DataOutputStream outToClient = new DataOutputStream(connectionSocket.getOutputStream());
                while (true) {
                    String str = "" + possiblePortTypes[rn.nextInt(possiblePortTypes.length)] + ","
                            + randomIps[rn.nextInt(numberOfRandomIps)] + ","
                            + randomIps[rn.nextInt(numberOfRandomIps)] + "\n";
                    outToClient.writeBytes(str);
                    Thread.sleep(10);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

    }
}

At the moment I have implemented the following function just to count the words, which i run with the following command in Mac OsX spark-submit spark_streaming.py <host> <port> <folder_name> <file_name>. I managed to establish the connection between the two and listening to the IPs generated. Now my main problem is how to keep track of the items I listen to.

from __future__ import print_function

import os
import sys

from pyspark import SparkContext
from pyspark.streaming import StreamingContext


# Get or register a Broadcast variable
def getWordBlacklist(sparkContext):
    if ('wordBlacklist' not in globals()):
        globals()['wordBlacklist'] = sparkContext.broadcast(["a", "b", "c"])
    return globals()['wordBlacklist']


# Get or register an Accumulator
def getDroppedWordsCounter(sparkContext):
    if ('droppedWordsCounter' not in globals()):
        globals()['droppedWordsCounter'] = sparkContext.accumulator(0)
    return globals()['droppedWordsCounter']


def createContext(host, port, outputPath):
    # If you do not see this printed, that means the StreamingContext has been loaded
    # from the new checkpoint
    print("Creating new context")
    if os.path.exists(outputPath):
        os.remove(outputPath)
    sc = SparkContext(appName="PythonStreamingRecoverableNetworkWordCount")
    ssc = StreamingContext(sc, 1)

    # Create a socket stream on target ip:port and count the
    # words in input stream of \n delimited text (eg. generated by 'nc')
    lines = ssc.socketTextStream(host, port)
    words = lines.flatMap(lambda line: line.split(" "))
    wordCounts = words.map(lambda x: (x, 1)).reduceByKey(lambda x, y: x + y)

    def echo(time, rdd):
        # Get or register the blacklist Broadcast
        blacklist = getWordBlacklist(rdd.context)
        # Get or register the droppedWordsCounter Accumulator
        droppedWordsCounter = getDroppedWordsCounter(rdd.context)

        # Use blacklist to drop words and use droppedWordsCounter to count them
        def filterFunc(wordCount):
            if wordCount[0] in blacklist.value:
                droppedWordsCounter.add(wordCount[1])
                return False
            else:
                return True

        counts = "Counts at time %s %s" % (time, rdd.filter(filterFunc).collect())
        print(counts)
        print("Dropped %d word(s) totally" % droppedWordsCounter.value)
        print("Appending to " + os.path.abspath(outputPath))
        # with open(outputPath, 'a') as f:
        #     f.write(counts + "\n")

    wordCounts.foreachRDD(echo)
    return ssc


if __name__ == "__main__":
    if len(sys.argv) != 5:
        print("Usage: recoverable_network_wordcount.py <hostname> <port> "
              "<checkpoint-directory> <output-file>", file=sys.stderr)
        sys.exit(-1)
    host, port, checkpoint, output = sys.argv[1:]
    ssc = StreamingContext.getOrCreate(checkpoint,
                                       lambda: createContext(host, int(port), output))
    ssc.start()
    ssc.awaitTermination()

At the end, I would like to read the stream and find the IP addresses per port that send or receive more than J packets in the last K seconds. J and K are some parameters I define in my code (like J=10 and K=60, etc.)

Answer 1

I have solved my problem using this method:

def getFrequentIps(stream, time_window, min_packets):
    frequent_ips = (stream.flatMap(lambda line: format_stream(line))            
                    # Count the occurrences of a specific pair 
                    .countByValueAndWindow(time_window, time_window, 4)
                    # Filter above the threshold imposed by min_packets
                    .filter(lambda count: count[1] >= int(min_packets))
                    .transform(lambda record: record.sortBy(lambda x: x[1], ascending=False)))

    number_items = 20
    print("Every %s seconds the top-%s channles with more than %s packages will be showed: " %
          (time_window, number_items, min_packets))
    frequent_ips.pprint(number_items)

Answer 2

As already mentioned by the answer you provided, PySpark has pre-built function that does exactly what you want, that is counting values over a time window.

countByValueAndWindow(windowLength, slideInterval, [numTasks]) 

Like in reduceByKeyAndWindow, the number of reduce tasks is configurable through an optional argument. Here you can find more examples: PySpark Documentation