- * 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 + * 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(); + /** + * @return the contents of a NiFi FlowFile + */ + byte[] getContent(); - /** - * Returns a Map of attributes that are associated with the NiFi FlowFile - * @return - */ - Map
- * 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.
+ * 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. + * 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. + * 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. *
- * + * *
*
+ * {@code
* 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
+ *
+ * // Create a JavaReceiverInputDStream using a NiFi receiver so that we can pull data from
* // specified Port
- * JavaReceiverInputDStream packetStream =
+ * 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() {
@@ -101,98 +104,105 @@ import org.apache.spark.streaming.receiver.Receiver;
* }
* }
* );
- *
+ *
* // print the results
* wordCounts.print();
* ssc.start();
* ssc.awaitTermination();
+ * }
*
*
*/
public class NiFiReceiver extends Receiver