You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/12/15 01:45:43 UTC
[GitHub] sijie closed pull request #3190: [Pulsar-Flink] Add Batch Sink
Scala Examples Documentation
sijie closed pull request #3190: [Pulsar-Flink] Add Batch Sink Scala Examples Documentation
URL: https://github.com/apache/pulsar/pull/3190
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/examples/flink-consumer-source/pom.xml b/examples/flink-consumer-source/pom.xml
index 3f407d58f1..c5e96b7ba8 100644
--- a/examples/flink-consumer-source/pom.xml
+++ b/examples/flink-consumer-source/pom.xml
@@ -73,7 +73,7 @@
<createDependencyReducedPom>false</createDependencyReducedPom>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
- <mainClass>org.apache.pulsar.examples.flink.PulsarConsumerSourceWordCount</mainClass>
+ <mainClass>org.apache.flink.streaming.connectors.pulsar.example.PulsarConsumerSourceWordCount</mainClass>
</transformer>
</transformers>
<finalName>pulsar-flink-streaming-wordcount</finalName>
diff --git a/pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkExample.java b/examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkExample.java
similarity index 100%
rename from pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkExample.java
rename to examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkExample.java
diff --git a/pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkExample.java b/examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkExample.java
similarity index 100%
rename from pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkExample.java
rename to examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkExample.java
diff --git a/pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkExample.java b/examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkExample.java
similarity index 100%
rename from pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkExample.java
rename to examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkExample.java
diff --git a/pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/README.md b/examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/README.md
similarity index 92%
rename from pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/README.md
rename to examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/README.md
index 2ab6ec0b6b..2f99e7622c 100644
--- a/pulsar-flink/src/test/java/org/apache/flink/batch/connectors/pulsar/example/README.md
+++ b/examples/flink-consumer-source/src/main/java/org/apache/flink/batch/connectors/pulsar/example/README.md
@@ -1,3 +1,24 @@
+<!--
+
+ 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.
+
+-->
+
The Flink Batch Sink for Pulsar is a custom sink that enables Apache [Flink](https://flink.apache.org/) to write [DataSet](https://ci.apache.org/projects/flink/flink-docs-stable/dev/batch/index.html) to Pulsar.
# Prerequisites
diff --git a/examples/flink-consumer-source/src/main/java/org/apache/pulsar/examples/flink/PulsarConsumerSourceWordCount.java b/examples/flink-consumer-source/src/main/java/org/apache/flink/streaming/connectors/pulsar/example/PulsarConsumerSourceWordCount.java
similarity index 98%
rename from examples/flink-consumer-source/src/main/java/org/apache/pulsar/examples/flink/PulsarConsumerSourceWordCount.java
rename to examples/flink-consumer-source/src/main/java/org/apache/flink/streaming/connectors/pulsar/example/PulsarConsumerSourceWordCount.java
index e163f603e1..547b635fb9 100644
--- a/examples/flink-consumer-source/src/main/java/org/apache/pulsar/examples/flink/PulsarConsumerSourceWordCount.java
+++ b/examples/flink-consumer-source/src/main/java/org/apache/flink/streaming/connectors/pulsar/example/PulsarConsumerSourceWordCount.java
@@ -16,7 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.examples.flink;
+package org.apache.flink.streaming.connectors.pulsar.example;
import static java.nio.charset.StandardCharsets.UTF_8;
diff --git a/examples/flink-consumer-source/README.md b/examples/flink-consumer-source/src/main/java/org/apache/flink/streaming/connectors/pulsar/example/README.md
similarity index 100%
rename from examples/flink-consumer-source/README.md
rename to examples/flink-consumer-source/src/main/java/org/apache/flink/streaming/connectors/pulsar/example/README.md
diff --git a/pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkScalaExample.scala b/examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkScalaExample.scala
similarity index 100%
rename from pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkScalaExample.scala
rename to examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkScalaExample.scala
diff --git a/pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkScalaExample.scala b/examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkScalaExample.scala
similarity index 100%
rename from pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkScalaExample.scala
rename to examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkScalaExample.scala
diff --git a/pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkScalaExample.scala b/examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkScalaExample.scala
similarity index 100%
rename from pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkScalaExample.scala
rename to examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkScalaExample.scala
diff --git a/examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/README.md b/examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/README.md
new file mode 100644
index 0000000000..b3f8cb5402
--- /dev/null
+++ b/examples/flink-consumer-source/src/main/scala/org/apache/flink/batch/connectors/pulsar/example/README.md
@@ -0,0 +1,283 @@
+<!--
+
+ 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.
+
+-->
+
+The Flink Batch Sink for Pulsar is a custom sink that enables Apache [Flink](https://flink.apache.org/) to write [DataSet](https://ci.apache.org/projects/flink/flink-docs-stable/dev/batch/index.html) to Pulsar.
+This document explains how to develop Scala Applications by using Flink Batch Sink.
+# Prerequisites
+
+To use this sink, include a dependency for the `pulsar-flink` library in your Java configuration.
+
+# Maven
+
+If you're using Maven, add this to your `pom.xml`:
+
+```xml
+<!-- in your <properties> block -->
+<pulsar.version>{{pulsar:version}}</pulsar.version>
+
+<!-- in your <dependencies> block -->
+<dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-flink</artifactId>
+ <version>${pulsar.version}</version>
+</dependency>
+```
+
+# Gradle
+
+If you're using Gradle, add this to your `build.gradle` file:
+
+```groovy
+def pulsarVersion = "{{pulsar:version}}"
+
+dependencies {
+ compile group: 'org.apache.pulsar', name: 'pulsar-flink', version: pulsarVersion
+}
+```
+
+# PulsarOutputFormat
+### Usage
+
+Please find Scala sample usage of `PulsarOutputFormat` as follows:
+
+```scala
+ /**
+ * Data type for words with count.
+ */
+ case class WordWithCount(word: String, count: Long) {
+ override def toString: String = "WordWithCount { word = " + word + ", count = " + count + " }"
+ }
+
+ /**
+ * Implementation
+ */
+ private val EINSTEIN_QUOTE = "Imagination is more important than knowledge. " +
+ "Knowledge is limited. Imagination encircles the world."
+ private val SERVICE_URL = "pulsar://127.0.0.1:6650"
+ private val TOPIC_NAME = "my-flink-topic"
+
+ def main(args: Array[String]): Unit = {
+
+ // set up the execution environment
+ val env = ExecutionEnvironment.getExecutionEnvironment
+
+ // create PulsarOutputFormat instance
+ val pulsarOutputFormat =
+ new PulsarOutputFormat[WordWithCount](SERVICE_URL, TOPIC_NAME, new SerializationSchema[WordWithCount] {
+ override def serialize(wordWithCount: WordWithCount): Array[Byte] = wordWithCount.toString.getBytes
+ })
+
+ // create DataSet
+ val textDS = env.fromElements[String](EINSTEIN_QUOTE)
+
+ // convert sentence to words
+ textDS.flatMap((value: String, out: Collector[WordWithCount]) => {
+ val words = value.toLowerCase.split(" ")
+ for (word <- words) {
+ out.collect(new WordWithCount(word.replace(".", ""), 1))
+ }
+ })
+
+ // filter words which length is bigger than 4
+ .filter((wordWithCount: WordWithCount) => wordWithCount.word.length > 4)
+
+ // group the words
+ .groupBy((wordWithCount: WordWithCount) => wordWithCount.word)
+
+ // sum the word counts
+ .reduce((wordWithCount1: WordWithCount, wordWithCount2: WordWithCount) =>
+ new WordWithCount(wordWithCount1.word, wordWithCount1.count + wordWithCount2.count))
+
+ // write batch data to Pulsar
+ .output(pulsarOutputFormat)
+
+ // set parallelism to write Pulsar in parallel (optional)
+ env.setParallelism(2)
+
+ // execute program
+ env.execute("Flink - Pulsar Batch WordCount")
+ }
+```
+
+### Sample Output
+
+Please find sample output for above application as follows:
+```
+WordWithCount { word = encircles, count = 1 }
+WordWithCount { word = important, count = 1 }
+WordWithCount { word = imagination, count = 2 }
+WordWithCount { word = limited, count = 1 }
+WordWithCount { word = knowledge, count = 2 }
+WordWithCount { word = world, count = 1 }
+```
+
+### Complete Example
+
+You can find a complete example [here](https://github.com/apache/incubator-pulsar/tree/master/pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchSinkScalaExample.scala).
+In this example, Flink DataSet is processed as word-count and being written to Pulsar.
+
+
+# PulsarCsvOutputFormat
+### Usage
+
+Please find Scala sample usage of `PulsarCsvOutputFormat` as follows:
+
+```scala
+ /**
+ * NasaMission Model
+ */
+ private case class NasaMission(id: Int, missionName: String, startYear: Int, endYear: Int)
+ extends Tuple4(id, missionName, startYear, endYear)
+
+ /**
+ * Implementation
+ */
+ private val SERVICE_URL = "pulsar://127.0.0.1:6650"
+ private val TOPIC_NAME = "my-flink-topic"
+
+ private val nasaMissions = List(
+ NasaMission(1, "Mercury program", 1959, 1963),
+ NasaMission(2, "Apollo program", 1961, 1972),
+ NasaMission(3, "Gemini program", 1963, 1966),
+ NasaMission(4, "Skylab", 1973, 1974),
+ NasaMission(5, "Apollo–Soyuz Test Project", 1975, 1975))
+
+ def main(args: Array[String]): Unit = {
+
+ // set up the execution environment
+ val env = ExecutionEnvironment.getExecutionEnvironment
+
+ // create PulsarCsvOutputFormat instance
+ val pulsarCsvOutputFormat =
+ new PulsarCsvOutputFormat[NasaMission](SERVICE_URL, TOPIC_NAME)
+
+ // create DataSet
+ val textDS = env.fromCollection(nasaMissions)
+
+ // map nasa mission names to upper-case
+ textDS.map(nasaMission => NasaMission(
+ nasaMission.id,
+ nasaMission.missionName.toUpperCase,
+ nasaMission.startYear,
+ nasaMission.endYear))
+
+ // filter missions which started after 1970
+ .filter(_.startYear > 1970)
+
+ // write batch data to Pulsar as Csv
+ .output(pulsarCsvOutputFormat)
+
+ // set parallelism to write Pulsar in parallel (optional)
+ env.setParallelism(2)
+
+ // execute program
+ env.execute("Flink - Pulsar Batch Csv")
+ }
+```
+
+### Sample Output
+
+Please find sample output for above application as follows:
+```
+4,SKYLAB,1973,1974
+5,APOLLO–SOYUZ TEST PROJECT,1975,1975
+```
+
+### Complete Example
+
+You can find a complete example [here](https://github.com/apache/incubator-pulsar/tree/master/pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchCsvSinkScalaExample.scala).
+In this example, Flink DataSet is processed and written to Pulsar in Csv format.
+
+
+# PulsarJsonOutputFormat
+### Usage
+
+Please find Scala sample usage of `PulsarJsonOutputFormat` as follows:
+
+```scala
+ /**
+ * NasaMission Model
+ */
+ private case class NasaMission(@BeanProperty id: Int,
+ @BeanProperty missionName: String,
+ @BeanProperty startYear: Int,
+ @BeanProperty endYear: Int)
+
+ /**
+ * Implementation
+ */
+ private val nasaMissions = List(
+ NasaMission(1, "Mercury program", 1959, 1963),
+ NasaMission(2, "Apollo program", 1961, 1972),
+ NasaMission(3, "Gemini program", 1963, 1966),
+ NasaMission(4, "Skylab", 1973, 1974),
+ NasaMission(5, "Apollo–Soyuz Test Project", 1975, 1975))
+
+ private val SERVICE_URL = "pulsar://127.0.0.1:6650"
+ private val TOPIC_NAME = "my-flink-topic"
+
+ def main(args: Array[String]): Unit = {
+
+ // set up the execution environment
+ val env = ExecutionEnvironment.getExecutionEnvironment
+
+ // create PulsarJsonOutputFormat instance
+ val pulsarJsonOutputFormat = new PulsarJsonOutputFormat[NasaMission](SERVICE_URL, TOPIC_NAME)
+
+ // create DataSet
+ val nasaMissionDS = env.fromCollection(nasaMissions)
+
+ // map nasa mission names to upper-case
+ nasaMissionDS.map(nasaMission =>
+ NasaMission(
+ nasaMission.id,
+ nasaMission.missionName.toUpperCase,
+ nasaMission.startYear,
+ nasaMission.endYear))
+
+ // filter missions which started after 1970
+ .filter(_.startYear > 1970)
+
+ // write batch data to Pulsar
+ .output(pulsarJsonOutputFormat)
+
+ // set parallelism to write Pulsar in parallel (optional)
+ env.setParallelism(2)
+
+ // execute program
+ env.execute("Flink - Pulsar Batch Json")
+ }
+```
+
+**Note:** Property definitions of the model should cover `@BeanProperty` to be visible.
+
+### Sample Output
+
+Please find sample output for above application as follows:
+```
+{"id":4,"missionName":"SKYLAB","startYear":1973,"endYear":1974}
+{"id":5,"missionName":"APOLLO–SOYUZ TEST PROJECT","startYear":1975,"endYear":1975}
+```
+
+### Complete Example
+
+You can find a complete example [here](https://github.com/apache/incubator-pulsar/tree/master/pulsar-flink/src/test/scala/org/apache/flink/batch/connectors/pulsar/example/FlinkPulsarBatchJsonSinkScalaExample.scala).
+In this example, Flink DataSet is processed and written to Pulsar in Json format.
diff --git a/pulsar-flink/pom.xml b/pulsar-flink/pom.xml
index e48d213aa6..b91af16ad3 100644
--- a/pulsar-flink/pom.xml
+++ b/pulsar-flink/pom.xml
@@ -51,12 +51,6 @@
<optional>true</optional>
</dependency>
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-scala_${scala.binary.version}</artifactId>
- <version>${flink.version}</version>
- </dependency>
-
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-client</artifactId>
diff --git a/site2/docs/io-rabbitmq.md b/site2/docs/io-rabbitmq.md
index d2cea153e7..0edaed186d 100644
--- a/site2/docs/io-rabbitmq.md
+++ b/site2/docs/io-rabbitmq.md
@@ -1,12 +1,12 @@
---
id: io-rabbitmq
title: RabbitMQ Connector
-sidebar_label: RabittMQ Connector
+sidebar_label: RabbitMQ Connector
---
## Source
-The RabittMQ Source connector is used for receiving messages from a RabittMQ cluster and writing
+The RabbitMQ Source connector is used for receiving messages from a RabbitMQ cluster and writing
messages to Pulsar topics.
### Source Configuration Options
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services