Repository: flink Updated Branches: refs/heads/master d353895ba -> 8227b0f75
[FLINK-3323] [docs] Add documentation for NiFi connector. This closes #2099 Project: http://git-wip-us.apache.org/repos/asf/flink/repo Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/8227b0f7 Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/8227b0f7 Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/8227b0f7 Branch: refs/heads/master Commit: 8227b0f7555b47e7eef268a81962908660212c3f Parents: d353895 Author: smarthi <smar...@apache.org> Authored: Mon May 30 09:04:25 2016 -0400 Committer: Fabian Hueske <fhue...@apache.org> Committed: Tue Jun 14 23:49:23 2016 +0200 ---------------------------------------------------------------------- docs/apis/streaming/connectors/elasticsearch.md | 2 +- .../apis/streaming/connectors/elasticsearch2.md | 2 +- docs/apis/streaming/connectors/index.md | 3 +- docs/apis/streaming/connectors/nifi.md | 128 +++++++++++++++++++ 4 files changed, 132 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/flink/blob/8227b0f7/docs/apis/streaming/connectors/elasticsearch.md ---------------------------------------------------------------------- diff --git a/docs/apis/streaming/connectors/elasticsearch.md b/docs/apis/streaming/connectors/elasticsearch.md index cf996fb..93b2bf6 100644 --- a/docs/apis/streaming/connectors/elasticsearch.md +++ b/docs/apis/streaming/connectors/elasticsearch.md @@ -180,4 +180,4 @@ text.addSink(new ElasticsearchSink(config, transports, new IndexRequestBuilder[S The difference is that we now need to provide a list of Elasticsearch Nodes to which the sink should connect using a `TransportClient`. -More about information about Elasticsearch can be found [here](https://elastic.co). +More information about Elasticsearch can be found [here](https://elastic.co). http://git-wip-us.apache.org/repos/asf/flink/blob/8227b0f7/docs/apis/streaming/connectors/elasticsearch2.md ---------------------------------------------------------------------- diff --git a/docs/apis/streaming/connectors/elasticsearch2.md b/docs/apis/streaming/connectors/elasticsearch2.md index 7146285..36d0920 100644 --- a/docs/apis/streaming/connectors/elasticsearch2.md +++ b/docs/apis/streaming/connectors/elasticsearch2.md @@ -141,4 +141,4 @@ This will buffer elements and Action Requests before sending to the cluster. The This now provides a list of Elasticsearch Nodes to which the sink should connect via a `TransportClient`. -More about information about Elasticsearch can be found [here](https://elastic.co). +More information about Elasticsearch can be found [here](https://elastic.co). http://git-wip-us.apache.org/repos/asf/flink/blob/8227b0f7/docs/apis/streaming/connectors/index.md ---------------------------------------------------------------------- diff --git a/docs/apis/streaming/connectors/index.md b/docs/apis/streaming/connectors/index.md index 20d6e41..85a07a1 100644 --- a/docs/apis/streaming/connectors/index.md +++ b/docs/apis/streaming/connectors/index.md @@ -4,7 +4,7 @@ title: "Streaming Connectors" # Sub-level navigation sub-nav-group: streaming sub-nav-id: connectors -sub-nav-pos: 6 +sub-nav-pos: 8 sub-nav-title: Connectors --- <!-- @@ -37,6 +37,7 @@ Currently these systems are supported: * [RabbitMQ](http://www.rabbitmq.com/) (sink/source) * [Amazon Kinesis Streams](http://aws.amazon.com/kinesis/streams/) (sink/source) * [Twitter Streaming API](https://dev.twitter.com/docs/streaming-apis) (source) + * [Apache NiFi](https://nifi.apache.org) (sink/source) To run an application using one of these connectors, additional third party components are usually required to be installed and launched, e.g. the servers http://git-wip-us.apache.org/repos/asf/flink/blob/8227b0f7/docs/apis/streaming/connectors/nifi.md ---------------------------------------------------------------------- diff --git a/docs/apis/streaming/connectors/nifi.md b/docs/apis/streaming/connectors/nifi.md new file mode 100644 index 0000000..fe7d2d3 --- /dev/null +++ b/docs/apis/streaming/connectors/nifi.md @@ -0,0 +1,128 @@ +--- +title: "Apache NiFi Connector" + +# Sub-level navigation +sub-nav-group: streaming +sub-nav-parent: connectors +sub-nav-pos: 7 +sub-nav-title: Apache NiFi +--- +<!-- +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. +--> + +This connector provides a Source and Sink that can read from and write to +[Apache NiFi](https://nifi.apache.org/). To use this connector, add the +following dependency to your project: + +{% highlight xml %} +<dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-connector-nifi{{ site.scala_version_suffix }}</artifactId> + <version>{{site.version }}</version> +</dependency> +{% endhighlight %} + +Note that the streaming connectors are currently not part of the binary +distribution. See +[here]({{site.baseurl}}/apis/cluster_execution.html#linking-with-modules-not-contained-in-the-binary-distribution) +for information about how to package the program with the libraries for +cluster execution. + +#### Installing Apache NiFi + +Instructions for setting up a Apache NiFi cluster can be found +[here](https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#how-to-install-and-start-nifi). + +#### Apache NiFi Source + +The connector provides a Source for reading data from Apache NiFi to Apache Flink. + +The class `NiFiSource(â¦)` provides 2 constructors for reading data from NiFi. + +- `NiFiSource(SiteToSiteConfig config)` - Constructs a `NiFiSource(â¦)` given the client's SiteToSiteConfig and a + default wait time of 1000 ms. + +- `NiFiSource(SiteToSiteConfig config, long waitTimeMs)` - Constructs a `NiFiSource(â¦)` given the client's + SiteToSiteConfig and the specified wait time (in milliseconds). + +Example: + +<div class="codetabs" markdown="1"> +<div data-lang="java" markdown="1"> +{% highlight java %} +SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder() + .url("http://localhost:8080/nifi") + .portName("Data for Flink") + .requestBatchCount(5) + .buildConfig(); + +SourceFunction<NiFiDataPacket> nifiSource = new NiFiSource(clientConfig); +{% endhighlight %} +</div> +<div data-lang="scala" markdown="1"> +{% highlight scala %} +val clientConfig: SiteToSiteClientConfig = new SiteToSiteClient.Builder() + .url("http://localhost:8080/nifi") + .portName("Data for Flink") + .requestBatchCount(5) + .buildConfig() + +val nifiSource = new NiFiSource(clientConfig) +{% endhighlight %} +</div> +</div> + +Here data is read from the Apache NiFi Output Port called "Data for Flink" which is part of Apache NiFi +Site-to_site protocol configuration. + +#### Apache NiFi Sink + +The connector provides a Sink for writing data from Apache NiFi to Apache Flink. + +The class `NiFiSink(â¦)` provides a constructor for instantiating a `NiFiSink`. `NiFiSink(SiteToSiteClientConfig, NiFiDataPacketBuilder<T>)` constructs a `NiFiSink(â¦)` given the client's `SiteToSiteConfig` and a `NiFiDataPacketBuilder` that converts data from Flink to `NiFiDataPacket` to be ingested by NiFi. + +Example: + +<div class="codetabs" markdown="1"> +<div data-lang="java" markdown="1"> +{% highlight java %} +SiteToSiteClientConfig clientConfig = new SiteToSiteClient.Builder() + .url("http://localhost:8080/nifi") + .portName("Data from Flink") + .requestBatchCount(5) + .buildConfig(); + +SinkFunction<NiFiDataPacket> nifiSink = new NiFiSink<>(clientConfig); +{% endhighlight %} +</div> +<div data-lang="scala" markdown="1"> +{% highlight scala %} +val clientConfig: SiteToSiteClientConfig = new SiteToSiteClient.Builder() + .url("http://localhost:8080/nifi") + .portName("Data from Flink") + .requestBatchCount(5) + .buildConfig() + +val nifiSink: NiFiSink[NiFiDataPacket] = new NiFiSink[NiFiDataPacket](clientConfig) +{% endhighlight %} +</div> +</div> + +More information about Apache NiFi Site-to-Site Protocol can be found [here](https://nifi.apache.org/docs/nifi-docs/html/user-guide.html#site-to-site) +More information about Apache NiFi can be found [here](https://nifi.apache.org). \ No newline at end of file