You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2015/02/19 03:20:06 UTC
incubator-nifi git commit: NIFI-365: Initial implementation of spark
receiver
Repository: incubator-nifi
Updated Branches:
refs/heads/nifi-site-to-site-client 7ab4392c1 -> 8506a0ce4
NIFI-365: Initial implementation of spark receiver
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/8506a0ce
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/8506a0ce
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/8506a0ce
Branch: refs/heads/nifi-site-to-site-client
Commit: 8506a0ce4ae71d2a103e11d6b8bf7e5832c81e09
Parents: 7ab4392
Author: Mark Payne <ma...@hotmail.com>
Authored: Wed Feb 18 21:19:58 2015 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Wed Feb 18 21:19:58 2015 -0500
----------------------------------------------------------------------
nifi/nifi-external/README.md | 19 ++
nifi/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(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/8506a0ce/nifi/nifi-external/README.md
----------------------------------------------------------------------
diff --git a/nifi/nifi-external/README.md b/nifi/nifi-external/README.md
new file mode 100644
index 0000000..649ad17
--- /dev/null
+++ b/nifi/nifi-external/README.md
@@ -0,0 +1,19 @@
+<!--
+ 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.
+-->
+# 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
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/8506a0ce/nifi/nifi-external/nifi-spark-receiver/pom.xml
----------------------------------------------------------------------
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 0000000..b21d554
--- /dev/null
+++ b/nifi/nifi-external/nifi-spark-receiver/pom.xml
@@ -0,0 +1,38 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ </parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-spark-receiver</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-streaming_2.10</artifactId>
+ <version>1.2.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-site-to-site-client</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ </dependency>
+ </dependencies>
+</project>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/8506a0ce/nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiDataPacket.java
----------------------------------------------------------------------
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 0000000..2f08dc5
--- /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;
+
+/**
+ * <p>
+ * 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
+ * </p>
+ */
+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<String, String> getAttributes();
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/8506a0ce/nifi/nifi-external/nifi-spark-receiver/src/main/java/org/apache/nifi/spark/NiFiReceiver.java
----------------------------------------------------------------------
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 0000000..9f31062
--- /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;
+
+
+/**
+ * <p>
+ * The <code>NiFiReceiver</code> 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.
+ * </p>
+ *
+ * <p>
+ * 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.
+ * </p>
+ *
+ * <p>
+ * 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.
+ * </p>
+ *
+ * <code>
+ * <pre>
+ * 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<NiFiDataPacket> packetStream =
+ * ssc.receiverStream(new NiFiReceiver(clientConfig, StorageLevel.MEMORY_ONLY()));
+ *
+ * // Map the data from NiFi to text, ignoring the attributes
+ * JavaDStream<String> text = packetStream.map(new Function<NiFiDataPacket, String>() {
+ * public String call(final NiFiDataPacket dataPacket) throws Exception {
+ * return new String(dataPacket.getContent(), StandardCharsets.UTF_8);
+ * }
+ * });
+ *
+ * // Split the words by spaces
+ * JavaDStream<String> words = text.flatMap(new FlatMapFunction<String, String>() {
+ * public Iterable<String> call(final String text) throws Exception {
+ * return Arrays.asList(SPACE.split(text));
+ * }
+ * });
+ *
+ * // Map each word to the number 1, then aggregate by key
+ * JavaPairDStream<String, Integer> wordCounts = words.mapToPair(
+ * new PairFunction<String, String, Integer>() {
+ * public Tuple2<String, Integer> call(String s) {
+ * return new Tuple2<String, Integer>(s, 1);
+ * }
+ * }).reduceByKey(new Function2<Integer, Integer, Integer>() {
+ * public Integer call(Integer i1, Integer i2) {
+ * return i1 + i2;
+ * }
+ * }
+ * );
+ *
+ * // print the results
+ * wordCounts.print();
+ * ssc.start();
+ * ssc.awaitTermination();
+ * </pre>
+ * </code>
+ */
+public class NiFiReceiver extends Receiver<NiFiDataPacket> {
+ 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<NiFiDataPacket> dataPackets = new ArrayList<NiFiDataPacket>();
+ 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<String, String> attributes = dataPacket.getAttributes();
+ final NiFiDataPacket NiFiDataPacket = new NiFiDataPacket() {
+ public byte[] getContent() {
+ return data;
+ }
+
+ public Map<String, String> 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);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/8506a0ce/nifi/nifi-external/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-external/pom.xml b/nifi/nifi-external/pom.xml
new file mode 100644
index 0000000..878098f
--- /dev/null
+++ b/nifi/nifi-external/pom.xml
@@ -0,0 +1,29 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi</artifactId>
+ <version>0.0.2-incubating-SNAPSHOT</version>
+ </parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-external</artifactId>
+ <packaging>pom</packaging>
+ <modules>
+ <module>nifi-spark-receiver</module>
+ </modules>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/8506a0ce/nifi/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/pom.xml b/nifi/pom.xml
index 6297161..5881db7 100644
--- a/nifi/pom.xml
+++ b/nifi/pom.xml
@@ -65,6 +65,7 @@
<module>nifi-assembly</module>
<module>nifi-docs</module>
<module>nifi-maven-archetypes</module>
+ <module>nifi-external</module>
</modules>
<scm>
<connection>scm:git:git://git.apache.org/incubator-nifi.git</connection>