I am using embedded spark and running Spark application with that along with nifi receiver for spark. On debugging I found that objects are created but no data comes there. # Site to Site properties nifi.remote.input.host=localhost nifi.remote.input.secure=false nifi.remote.input.socket.port=10000 nifi.remote.input.http.enabled=true nifi.remote.input.http.transaction.ttl=30 sec nifi.web.http.port=8080 Below is the code : SiteToSiteClientConfig config = new SiteToSiteClient.Builder()
.url("http://localhost:8080/nifi")
.portName("SparkPort")
.buildConfig(); SparkConf sparkConf = new SparkConf()
.setMaster("local[2]")
.setAppName("SparkNifiConnectorApp"); JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(1)); // Create a JavaReceiverInputDStream using a NiFi receiver so that we can pull data from
// specified Port while(true) { JavaReceiverInputDStream packetStream =
ssc.receiverStream(new NiFiReceiver(config, StorageLevel.MEMORY_ONLY())); // Map the data from NiFi to text, ignoring the attributes System.out.println("Packet Content = " + packetStream); System.out.println("Packet Count = " + packetStream.countByValue()); System.out.println("Packet Count = " + packetStream.count()); JavaDStream text = packetStream.map(new Function<NiFiDataPacket, String>() { public String call(final NiFiDataPacket dataPacket) throws Exception {
System.out.println("Packet Data : " +dataPacket.getContent()); return new String(dataPacket.getContent(), StandardCharsets.UTF_8); } }); System.out.println("Data from Nifi :" + text.toString()); text.print();
... View more
