From 8506a0ce4ae71d2a103e11d6b8bf7e5832c81e09 Mon Sep 17 00:00:00 2001 From: Mark Payne Date: Wed, 18 Feb 2015 21:19:58 -0500 Subject: [PATCH] NIFI-365: Initial implementation of spark receiver --- nifi/nifi-external/README.md | 19 ++ .../nifi-external/nifi-spark-receiver/pom.xml | 38 ++++ .../org/apache/nifi/spark/NiFiDataPacket.java | 40 ++++ .../org/apache/nifi/spark/NiFiReceiver.java | 198 ++++++++++++++++++ nifi/nifi-external/pom.xml | 29 +++ nifi/pom.xml | 1 + 6 files changed, 325 insertions(+) create mode 100644 nifi/nifi-external/README.md create mode 100644 nifi/nifi-external/nifi-spark-receiver/pom.xml create mode 100644 nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiDataPacket.java create mode 100644 nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java create mode 100644 nifi/nifi-external/pom.xml diff --git a/nifi/nifi-external/README.md b/nifi/nifi-external/README.md new file mode 100644 index 0000000000..649ad176b8 --- /dev/null +++ b/nifi/nifi-external/README.md @@ -0,0 +1,19 @@ + +# nifi-external + +The nifi-external module is a location where components can be developed by the NiFi team +that are not intended to be used directly by NiFi but are to be used within other frameworks +in order to integrate with NiFi. \ No newline at end of file diff --git a/nifi/nifi-external/nifi-spark-receiver/pom.xml b/nifi/nifi-external/nifi-spark-receiver/pom.xml new file mode 100644 index 0000000000..b21d554fbc --- /dev/null +++ b/nifi/nifi-external/nifi-spark-receiver/pom.xml @@ -0,0 +1,38 @@ + + + + 4.0.0 + + org.apache.nifi + nifi + 0.0.2-incubating-SNAPSHOT + + org.apache.nifi + nifi-spark-receiver + + + + org.apache.spark + spark-streaming_2.10 + 1.2.0 + + + org.apache.nifi + nifi-site-to-site-client + 0.0.2-incubating-SNAPSHOT + + + \ No newline at end of file diff --git a/nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiDataPacket.java b/nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiDataPacket.java new file mode 100644 index 0000000000..2f08dc5c6d --- /dev/null +++ b/nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiDataPacket.java @@ -0,0 +1,40 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.spark; + +import java.util.Map; + +/** + *

+ * The NiFiDataPacket provides a packaging around a NiFi FlowFile. It wraps both a FlowFile's + * content and its attributes so that they can be processed by Spark + *

+ */ +public interface NiFiDataPacket { + + /** + * Returns the contents of a NiFi FlowFile + * @return + */ + byte[] getContent(); + + /** + * Returns a Map of attributes that are associated with the NiFi FlowFile + * @return + */ + Map getAttributes(); +} diff --git a/nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java b/nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java new file mode 100644 index 0000000000..9f3106210c --- /dev/null +++ b/nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java @@ -0,0 +1,198 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.spark; + +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import org.apache.nifi.remote.Transaction; +import org.apache.nifi.remote.TransferDirection; +import org.apache.nifi.remote.client.SiteToSiteClient; +import org.apache.nifi.remote.client.SiteToSiteClientConfig; +import org.apache.nifi.remote.protocol.DataPacket; +import org.apache.nifi.stream.io.StreamUtils; +import org.apache.spark.storage.StorageLevel; +import org.apache.spark.streaming.receiver.Receiver; + + +/** + *

+ * The NiFiReceiver is a Reliable Receiver that provides a way to pull data + * from Apache NiFi so that it can be processed by Spark Streaming. The NiFi Receiver connects + * to NiFi instance provided in the config and requests data from + * the OutputPort that is named. In NiFi, when an OutputPort is added to the root process group, + * it acts as a queue of data for remote clients. This receiver is then able to pull that data + * from NiFi reliably. + *

+ * + *

+ * It is important to note that if pulling data from a NiFi cluster, the URL that should be used + * is that of the NiFi Cluster Manager. The Receiver will automatically handle determining the nodes + * in that cluster and pull from those nodes as appropriate. + *

+ * + *

+ * In order to use the NiFiReceiver, you will need to first build a {@link SiteToSiteClientConfig} to provide + * to the constructor. This can be achieved by using the {@link SiteToSiteClient.Builder}. + * Below is an example snippet of driver code to pull data from NiFi that is running on localhost:8080. This + * example assumes that NiFi exposes and OutputPort on the root group named "Data For Spark". + * Additionally, it assumes that the data that it will receive from this OutputPort is text + * data, as it will map the byte array received from NiFi to a UTF-8 Encoded string. + *

+ * + * + * 
+ * Pattern SPACE = Pattern.compile(" ");
+ * 
+ * // Build a Site-to-site client config
+ * SiteToSiteClientConfig config = new SiteToSiteClient.Builder()
+ *   .setUrl("http://localhost:8080/nifi")
+ *   .setPortName("Data For Spark")
+ *   .buildConfig();
+ * 
+ * SparkConf sparkConf = new SparkConf().setAppName("NiFi-Spark Streaming example");
+ * JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, new Duration(1000L));
+ * 
+ * // Create a JavaReceiverInputDStream using a NiFi receiver so that we can pull data from 
+ * // specified Port
+ * JavaReceiverInputDStream packetStream = 
+ *     ssc.receiverStream(new NiFiReceiver(clientConfig, StorageLevel.MEMORY_ONLY()));
+ * 
+ * // Map the data from NiFi to text, ignoring the attributes
+ * JavaDStream text = packetStream.map(new Function() {
+ *   public String call(final NiFiDataPacket dataPacket) throws Exception {
+ *     return new String(dataPacket.getContent(), StandardCharsets.UTF_8);
+ *   }
+ * });
+ * 
+ * // Split the words by spaces
+ * JavaDStream words = text.flatMap(new FlatMapFunction() {
+ *   public Iterable call(final String text) throws Exception {
+ *     return Arrays.asList(SPACE.split(text));
+ *   }
+ * });
+ * 	    
+ * // Map each word to the number 1, then aggregate by key
+ * JavaPairDStream wordCounts = words.mapToPair(
+ *   new PairFunction() {
+ *     public Tuple2 call(String s) {
+ *       return new Tuple2(s, 1);
+ *     }
+ *   }).reduceByKey(new Function2() {
+ *     public Integer call(Integer i1, Integer i2) {
+ *       return i1 + i2;
+ *     }
+ *    }
+ *  );
+ * 
+ * // print the results
+ * wordCounts.print();
+ * ssc.start();
+ * ssc.awaitTermination();
+ *
+ * + */ +public class NiFiReceiver extends Receiver { + private static final long serialVersionUID = 3067274587595578836L; + private final SiteToSiteClientConfig clientConfig; + + public NiFiReceiver(final SiteToSiteClientConfig clientConfig, final StorageLevel storageLevel) { + super(storageLevel); + this.clientConfig = clientConfig; + } + + @Override + public void onStart() { + final Thread thread = new Thread(new ReceiveRunnable()); + thread.setDaemon(true); + thread.setName("NiFi Receiver"); + thread.start(); + } + + @Override + public void onStop() { + } + + class ReceiveRunnable implements Runnable { + public ReceiveRunnable() { + } + + public void run() { + try { + final SiteToSiteClient client = new SiteToSiteClient.Builder().fromConfig(clientConfig).build(); + try { + while ( !isStopped() ) { + final Transaction transaction = client.createTransaction(TransferDirection.RECEIVE); + DataPacket dataPacket = transaction.receive(); + if ( dataPacket == null ) { + transaction.confirm(); + transaction.complete(); + + // no data available. Wait a bit and try again + try { + Thread.sleep(1000L); + } catch (InterruptedException e) {} + + continue; + } + + final List dataPackets = new ArrayList(); + do { + // Read the data into a byte array and wrap it along with the attributes + // into a NiFiDataPacket. + final InputStream inStream = dataPacket.getData(); + final byte[] data = new byte[(int) dataPacket.getSize()]; + StreamUtils.fillBuffer(inStream, data); + + final Map attributes = dataPacket.getAttributes(); + final NiFiDataPacket NiFiDataPacket = new NiFiDataPacket() { + public byte[] getContent() { + return data; + } + + public Map getAttributes() { + return attributes; + } + }; + + dataPackets.add(NiFiDataPacket); + dataPacket = transaction.receive(); + } while ( dataPacket != null ); + + // Confirm transaction to verify the data + transaction.confirm(); + + store(dataPackets.iterator()); + + transaction.complete(); + } + } finally { + try { + client.close(); + } catch (final IOException ioe) { + reportError("Failed to close client", ioe); + } + } + } catch (final IOException ioe) { + restart("Failed to receive data from NiFi", ioe); + } + } + } +} diff --git a/nifi/nifi-external/pom.xml b/nifi/nifi-external/pom.xml new file mode 100644 index 0000000000..878098f4a6 --- /dev/null +++ b/nifi/nifi-external/pom.xml @@ -0,0 +1,29 @@ + + + + 4.0.0 + + org.apache.nifi + nifi + 0.0.2-incubating-SNAPSHOT + + org.apache.nifi + nifi-external + pom + + nifi-spark-receiver + + diff --git a/nifi/pom.xml b/nifi/pom.xml index 62971611a4..5881db7a03 100644 --- a/nifi/pom.xml +++ b/nifi/pom.xml @@ -65,6 +65,7 @@ nifi-assembly nifi-docs nifi-maven-archetypes + nifi-external scm:git:git://git.apache.org/incubator-nifi.git