在SPARK中实现接收器

Question

我一直在试图为SPARK 0.9实现接收器。我使用Jnetpcap库捕获数据包，并需要将它传递给Scala中的Spark。在“def receive（）”方法中写入数据包的捕获部分足够了吗？

编辑：这里是this link代码使用Jnetpcap库捕获数据包：

import java.util.Date 
import java.lang.StringBuilder 
import org.jnetpcap.Pcap 
import org.jnetpcap.packet.PcapPacket 
import org.jnetpcap.packet.PcapPacketHandler 

object PacketCapture1 { 
    def main(args: Array[String]){ 
    val snaplen = 64 * 1024 // Capture all packets, no trucation 
    val flags = Pcap.MODE_PROMISCUOUS // capture all packets 
    val timeout = 10 * 1000 
    //val errbuf = new StringBuilder() 

    val jsb = new java.lang.StringBuilder() 
    val errbuf = new StringBuilder(jsb); 
    val pcap = Pcap.openLive("eth0", snaplen, flags, timeout, errbuf) 
    if (pcap == null) { 
     println("Error : " + errbuf.toString()) 
    } 
    println(pcap) 
    val jpacketHandler = new PcapPacketHandler[String]() { 

     def nextPacket(packet: PcapPacket, user: String) { 
     println("Received packet at %s caplen=%4d len=%4d %s\n", new Date(packet.getCaptureHeader.timestampInMillis()), 
      packet.getCaptureHeader.caplen(), packet.getCaptureHeader.wirelen(), user) 
     } 
    } 
    pcap.loop(30, jpacketHandler, "jNetPcap works!") 
    pcap.close() 

    } 
} 

如何实现使用此代码捕获的数据包接收器的火花？

Answer 1

您必须创建一个自定义NetworkReceiver（或Spark 1.0+中的Receiver）并实施onStart()方法。对于星火0.9，看到http://spark.apache.org/docs/0.9.1/streaming-custom-receivers.html

火花1.0（强烈推荐），见http://spark.apache.org/docs/latest/streaming-custom-receivers.html