Implement generic receiver model with the integration of spark streaming about openxdr HOT 4 CLOSED

package JavaRecieverTest
import com.google.common.io.Closeables;

import org.apache.spark.SparkConf;
import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.receiver.Receiver;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.ConnectException;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

/**

• Implement Receiver for any customer resources
  */
  public class TestJavaReceiver extends Receiver{
  public static void main(String[] args){
  if(args.length < 2){
  System.err.println("Usage: TestJavaReceiver ");
  System.exit(1);
  }
  }

  //create context
  SparkConf sparkConf = new SparkConf().setAppName("TestJavaReceiver");
  JavaStreamingContext jsc = new JavaStreamingContext(sparkConf, new Duration(1000));//1 second

  //Create a receiver inputstream based on provided IP port
  //lines is the input data from any resources
  JavaReceiverInputDStream lines = jsc.receiverStream(
  new TestJavaReceiver(args[0], Integer.parseInt(args[1]))
  ); //A Java-friendly interface to ReceiverInputDStream, the abstract class for defining any input stream that receives data over the network.

  /**After receiving data, we could do a lot of thing here.

  • Such as map them;
  • and process them.
    */

  //define the input originals
  String host = null;
  int port = -1;

  public TestJavaReceiver(String host, int port){
  super(StorageLevel.MEMORY_AND_DISK_SER());//replicate data into 2 cluster seems useless
  this.host = host;
  this.port = port;
  }

  /**

  • Start threads when receiving data over a connection
    */
    public void onStart(){
    new Thread(){
    public void run(){
    receive();
    }
    }.start();
    }

  //The thread will stop itself in receive() function
  // public void onStop() {}

  /**

  • Create a socket connection and receive data streams
    /
    private void receive(){
    Socket socket = null;
    BufferedReader reader = null;
    String userInput = null;
    try{
    try{
    //connect to data sources
    socket = new Socket(host, port);
    //read data sources as UTF-8 charset
    reader = new BufferReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
    //Until stopped or connection broken down
    while(!isStopped() && (userInput = reader.readLine()) != null){
    store(userInput);//Receiver's built-in function
    }
    }finally{
    Closeables.close(reader, / swallowIOException / true);
    Closeables.close(socket, / swallowIOException */ true);
    }
    //restart to try again
    restart("try to connect sources again");
    }catch(ConnectException cex){
    //restart connection
    restart("Could not connect data source", cex);
    }catch(Throwable t){
    //restart connection
    restart("Error receiving data",t);
    }
    }

  public JavaReceiverInputDStream getData(){
  return this.lines;
  }
  }

from openxdr.

Hi, after finishing my thesis. I wrote one example here. The receiver retrieve data sources from user's defined ip and port. It originates from Spark document:http://spark.apache.org/docs/latest/streaming-custom-receivers.html. But I made some changes.

from openxdr.

Yes, the abstract implementation should depend on spark custom receivers, You can extend the basic 'Receiver' to support various data sources OpenADS support.

from openxdr.

@xiaoleihuang Please add this code under 'feature/receiver' branch and send a pull-request to merge to develop, then merge it with 'feature/receiver-pcap' and develop your pcap code

from openxdr.