You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by pm...@apache.org on 2018/10/19 19:49:34 UTC
samza git commit: Eventhubs 1.0 tutorials cleanup
Repository: samza
Updated Branches:
refs/heads/master 3523122af -> 5ba543faf
Eventhubs 1.0 tutorials cleanup
Misc Event hubs cleanup for the 1.0 release:
- Updated hello-samza tutorial documentation
- Removed unused depricated documentation (all content is not in `samza-event-hubs-standalone.md` or `connectors/eventhubs.md`
- Fixed some integration tests
- Added Event Hubs tutorial in the tutorials `index.md`
Screenshots:
<img width="1680" alt="screen shot 2018-10-16 at 11 56 51 pm" src="https://user-images.githubusercontent.com/29577458/47067812-62a84400-d19f-11e8-9511-2a3e652f4b75.png">
<img width="1680" alt="screen shot 2018-10-16 at 11 57 09 pm" src="https://user-images.githubusercontent.com/29577458/47067822-66d46180-d19f-11e8-8121-0f6fe09cbce1.png">
<img width="1680" alt="screen shot 2018-10-16 at 11 57 19 pm" src="https://user-images.githubusercontent.com/29577458/47067826-6936bb80-d19f-11e8-80a1-0e94e8459468.png">
<img width="1680" alt="screen shot 2018-10-16 at 11 57 31 pm" src="https://user-images.githubusercontent.com/29577458/47067828-6a67e880-d19f-11e8-9828-b672d0b3a1a0.png">
vjagadish prateekm please review
Author: Daniel Chen <dc...@linkedin.com>
Reviewers: Jagadish Venkatraman <vj...@gmail.com>
Closes #736 from dxichen/eventhubs-tutorial-cleanup
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/5ba543fa
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/5ba543fa
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/5ba543fa
Branch: refs/heads/master
Commit: 5ba543faf1ce68d1398ebc4ce2da72f70149e576
Parents: 3523122
Author: Daniel Chen <dc...@linkedin.com>
Authored: Fri Oct 19 12:49:24 2018 -0700
Committer: Prateek Maheshwari <pm...@apache.org>
Committed: Fri Oct 19 12:49:24 2018 -0700
----------------------------------------------------------------------
.../azure/eventhub_send_methods.png | Bin 52427 -> 0 bytes
.../documentation/versioned/azure/eventhubs.md | 201 -------------------
.../versioned/connectors/eventhubs.md | 121 ++++++++++-
docs/learn/tutorials/versioned/index.md | 4 +-
.../versioned/samza-event-hubs-standalone.md | 127 ++++++++++++
.../consumer/ITestEventHubSystemConsumer.java | 2 +-
.../producer/ITestEventHubSystemProducer.java | 2 +-
7 files changed, 244 insertions(+), 213 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/5ba543fa/docs/learn/documentation/azure/eventhub_send_methods.png
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/azure/eventhub_send_methods.png b/docs/learn/documentation/azure/eventhub_send_methods.png
deleted file mode 100644
index 477553a..0000000
Binary files a/docs/learn/documentation/azure/eventhub_send_methods.png and /dev/null differ
http://git-wip-us.apache.org/repos/asf/samza/blob/5ba543fa/docs/learn/documentation/versioned/azure/eventhubs.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/azure/eventhubs.md b/docs/learn/documentation/versioned/azure/eventhubs.md
deleted file mode 100644
index 18e9ebd..0000000
--- a/docs/learn/documentation/versioned/azure/eventhubs.md
+++ /dev/null
@@ -1,201 +0,0 @@
----
-layout: page
-title: Connecting to Event Hubs
----
-<!--
- 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.
--->
-
-You can configure your Samza jobs to process data from [Azure Event Hubs](https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-features), Microsoft's data streaming service. An `event hub` is similar to a Kafka topic and can have multiple partitions with producers and consumers. Each message produced or consumed from an event hub is an instance of [EventData](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data).
-
-### Consuming from Event Hubs:
-
-Samza's [EventHubSystemConsumer](https://github.com/apache/samza/blob/master/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java) wraps the EventData into an [EventHubIncomingMessageEnvelope](https://github.com/apache/samza/blob/master/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubIncomingMessageEnvelope.java). The key of the message is set to the partition key of the EventData. The message is obtained from the EventData body.
-
-To configure Samza to configure from Event Hubs streams:
-
-```
-# define an event hub system factory with your identifier. eg: eh-system
-systems.eh-system.samza.factory=org.apache.samza.system.eventhub.EventHubSystemFactory
-
-# define your streams
-systems.eh-system.stream.list=input0, output0
-
-# define required properties for your streams
-systems.eh-system.streams.input0.eventhubs.namespace=YOUR-STREAM-NAMESPACE
-systems.eh-system.streams.input0.eventhubs.entitypath=YOUR-ENTITY-NAME
-systems.eh-system.streams.input0.eventhubs.sas.keyname=YOUR-SAS-KEY-NAME
-systems.eh-system.streams.input0.eventhubs.sas.token=YOUR-SAS-KEY-TOKEN
-
-systems.eh-system.streams.output0.eventhubs.namespace=YOUR-STREAM-NAMESPACE
-systems.eh-system.streams.output0.eventhubs.entitypath=YOUR-ENTITY-NAME
-systems.eh-system.streams.output0.eventhubs.sas.keyname=YOUR-SAS-KEY-NAME
-systems.eh-system.streams.output0.eventhubs.sas.token=YOUR-SAS-KEY-TOKEN
-```
-
-The tuple required to access the Event Hubs entity per stream must be provided, namely the fields `YOUR-STREAM-NAMESPACE`, `YOUR-ENTITY-NAME`, `YOUR-SAS-KEY-NAME`, `YOUR-SAS-KEY-TOKEN`.
-
-### Producing to Event Hubs:
-
-Similarly, you can also configure your Samza job to write to Event Hubs.
-```
-OutgoingMessageEnvelope envelope = new OutgoingMessageEnvelope(new SystemStream("eh-system", "output0"), key, message);
-collector.send(envelope);
-```
-
-Each [OutgoingMessageEnvelope](https://samza.apache.org/learn/documentation/latest/api/javadocs/org/apache/samza/system/OutgoingMessageEnvelope.html) is converted into an [EventData](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data) instance whose body is set to the `message` in the envelope. Additionally, the `key` and the `produce timestamp` are set as properties in the EventData before sending it to Event Hubs.
-
-#### Size limit of partition key:
-
-Note that Event Hubs has a limit on the length of partition key (128 characters). In [EventHubSystemProducer](https://github.com/apache/samza/blob/master/samza-azure/src/main/java/org/apache/samza/system/eventhub/producer/EventHubSystemProducer.java) we truncate the partition key if the size of the key exceeds the limit.
-
-### Advanced configuration:
-
-##### Producer partitioning:
-
-The `partition.method` property determines how outgoing messages are partitioned. Valid values for this config are `EVENT_HUB_HASHING`, `PARTITION_KEY_AS_PARTITION` or `ROUND_ROBIN`.
-
-`EVENT_HUB_HASHING`: By default, Samza computes the partition for an outgoing message based on the hash of its partition-key. This ensures that events with the same key are sent to the same partition. If this option is chosen, the partition key should be a string. If the partition key is not set, the key in the message is used for partitioning.
-
-`PARTITION_KEY_AS_PARTITION`: In this method, each message is sent to the partition specified by its partition key. This requires the partition key to be an integer. If the key is greater than the number of partitions, a modulo operation will be performed on the key. Similar to EVENT_HUB_HASHING, the key in the message is used if the partition key is not specified.
-
-`ROUND_ROBIN`: In this method, outgoing messages are distributed in a round-robin across all partitions. The key and the partition key in the message are ignored.
-
-
-
-```
-systems.eh-system.partition.method = EVENT_HUB_HASHING
-```
-
-##### Consumer groups:
-
-Event Hubs supports a notion of [consumer groups](https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-features#consumer-groups) which enable multiple applications have their own view of the event stream. Each event hub stream has a pre-defined consumer group named `$Default`. You can define your own consumer group for your job and configure a `eventhubs.consumer.group`
-
-```
-systems.eh-system.streams.eh-input0.eventhubs.consumer.group = my-group
-```
-
-##### Serde:
-
-By default, the messages from Event Hubs are sent and received as byte arrays. You can configure a serializer and deserializer for your message by setting a value for `msg.serde` for your stream.
-
-```
-streams.input0.samza.msg.serde = json
-streams.output0.samza.msg.serde = json
-```
-
-##### Consumer buffer size:
-
-When the consumer reads a message from event hubs, it appends them to a shared producer-consumer buffer corresponding to its partition. This config determines the per-partition queue size. Setting a higher value for this config typically achieves a higher throughput at the expense of increased on-heap memory.
-
-```
-systems.eh-system.eventhubs.receive.queue.size = 10
-```
-
-For the list of all configs, check out the configuration table page [here](../jobs/configuration-table.html)
-
-### Azure Event Hubs Hello-Samza Example
-
-The [hello-samza](https://github.com/apache/samza-hello-samza) project contains an example of a high level job that consumes and produces to Event Hubs using the Zookeeper deployment model.
-
-#### Get the Code
-
-Let's get started by cloning the hello-samza project
-
-```
-git clone https://git.apache.org/samza-hello-samza.git hello-samza
-cd hello-samza
-git checkout latest
-```
-
-The project comes up with numerous examples and for this tutorial, we will pick the Azure application.
-
-#### Setting up the Deployment Environment
-
-For our Azure application, we require [ZooKeeper](http://zookeeper.apache.org/). The hello-samza project comes with a script called "grid" to help with the environment setup
-
-```
-./bin/grid standalone
-```
-
-This command will download, install, and start ZooKeeper and Kafka. It will also check out the latest version of Samza and build it. All package files will be put in a sub-directory called "deploy" inside hello-samza's root folder.
-
-If you get a complaint that JAVA_HOME is not set, then you'll need to set it to the path where Java is installed on your system.
-
-
-#### Configuring the Azure application
-
-Here are the configs you must set before building the project. Configure these in the `src/main/config/azure-application-local-runner.properties` file.
-
-```
-# Add your Event Hubs input stream credentials here
-systems.eventhubs.streams.input-stream.eventhubs.namespace=YOUR-STREAM-NAMESPACE
-systems.eventhubs.streams.input-stream.eventhubs.entitypath=YOUR-ENTITY-NAME
-systems.eventhubs.streams.input-stream.eventhubs.sas.keyname=YOUR-SAS-KEY-NAME
-systems.eventhubs.streams.input-stream.eventhubs.sas.token=YOUR-SAS-KEY-TOKEN
-
-# Add your Event Hubs output stream credentials here
-systems.eventhubs.streams.output-stream.eventhubs.namespace=YOUR-STREAM-NAMESPACE
-systems.eventhubs.streams.output-stream.eventhubs.entitypath=YOUR-ENTITY-NAME
-systems.eventhubs.streams.output-stream.eventhubs.sas.keyname=YOUR-SAS-KEY-NAME
-systems.eventhubs.streams.output-stream.eventhubs.sas.token=YOUR-SAS-KEY-TOKEN
-```
-
-Optionally, you may also use the Azure Checkpoint Manager. Otherwise, comment out both these lines.
-
-```
-# Azure Table Checkpoint Manager
-task.checkpoint.factory=org.apache.samza.checkpoint.azure.AzureCheckpointManagerFactory
-azure.storage.connect=YOUR-STORAGE-ACCOUNT-CONNECTION-STRING
-```
-
-#### Building the Hello Samza Project
-
-With the environment setup complete, let us move on to building the hello-samza project. Execute the following commands:
-
-```
-mvn clean package
-mkdir -p deploy/samza
-tar -xvf ./target/hello-samza-0.15.0-SNAPSHOT-dist.tar.gz -C deploy/samza
-```
-
-We are now all set to deploy the application locally.
-
-#### Running the Azure application
-
-In order to run the application, we will use the *run-azure-application* script.
-
-```
-./deploy/samza/bin/run-azure-application.sh
-```
-
-The above command executes the helper script which invokes the *AzureZKLocalApplication* main class, which starts the *AzureApplication*. This application filters out the messages consumed without keys, prints them out and send them the configured output stream.
-
-The messages consumed should be printed in the following format:
-```
-Sending:
-Received Key: <KEY>
-Received Message: <VALUE>
-```
-
-#### Shutdown
-
-This application can be shutdown by terminating the *run-azure-application* script.
-We can use the *grid* script to tear down the local environment ([Kafka](http://kafka.apache.org/) and [Zookeeper](http://zookeeper.apache.org/)).
-
-```
-bin/grid stop all
-```
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/5ba543fa/docs/learn/documentation/versioned/connectors/eventhubs.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/connectors/eventhubs.md b/docs/learn/documentation/versioned/connectors/eventhubs.md
index e805efb..16120f3 100644
--- a/docs/learn/documentation/versioned/connectors/eventhubs.md
+++ b/docs/learn/documentation/versioned/connectors/eventhubs.md
@@ -21,13 +21,49 @@ title: Event Hubs Connector
## Overview
-The Samza Event Hubs connector provides access to [Azure Event Hubs](https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-features), Microsoft’s data streaming service on Azure. An event hub is similar to a Kafka topic and can have multiple partitions with producers and consumers. Each message produced or consumed from an event hub is an instance of [EventData](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data).
+The Samza Event Hubs connector provides access to [Azure Event Hubs](https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-features), Microsoft’s data streaming service on Azure. An event hub is similar to a Kafka topic and can have multiple partitions with producers and consumers. Each message produced or consumed from an event hub is an instance of [EventData](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data).
+
+You may find an [example](../../../tutorials/versioned/samza-event-hubs-standalone.md) using this connector in the [hello-samza](https://github.com/apache/samza-hello-samza) project.
## Consuming from Event Hubs
-Samza’s [EventHubSystemConsumer](https://github.com/apache/samza/blob/master/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java) wraps the EventData into an [EventHubIncomingMessageEnvelope](https://github.com/apache/samza/blob/master/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubIncomingMessageEnvelope.java). Samza's Event Hubs consumer wraps each message from Event Hubs into an EventHubMessageEnvelope. The envelope has two fields of interest - the key, which is set to the event's partition key and the message, which is set to the actual data in the event.
+Samza’s [EventHubSystemConsumer](https://github.com/apache/samza/blob/master/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubSystemConsumer.java) wraps the EventData into an [EventHubIncomingMessageEnvelope](https://github.com/apache/samza/blob/master/samza-azure/src/main/java/org/apache/samza/system/eventhub/consumer/EventHubIncomingMessageEnvelope.java). Samza's Event Hubs consumer wraps each message from Event Hubs into an EventHubMessageEnvelope. The envelope has two fields of interest - the key, which is set to the event's String partition key and the message, which is set to the actual data in the event.
+
+You can describe your Samza jobs to process data from Azure Event Hubs. To set Samza to consume from Event Hubs streams:
+
+{% highlight java %}
+ 1 public void describe(StreamApplicationDescriptor appDescriptor) {
+ 2 // Define your system here
+ 3 EventHubsSystemDescriptor systemDescriptor = new EventHubsSystemDescriptor("eventhubs");
+ 4
+ 5 // Choose your serializer/deserializer for the EventData payload
+ 6 StringSerde serde = new StringSerde();
+ 7
+ 8 // Define the input descriptors with respective descriptors
+ 9 EventHubsInputDescriptor<KV<String, String>> inputDescriptor =
+10 systemDescriptor.getInputDescriptor(INPUT_STREAM_ID, EVENTHUBS_NAMESPACE, EVENTHUBS_INPUT_ENTITY, serde)
+11 .withSasKeyName(EVENTHUBS_SAS_KEY_NAME)
+12 .withSasKey(EVENTHUBS_SAS_KEY_TOKEN);
+13
+14 // Define the input streams with descriptors
+15 MessageStream<KV<String, String>> eventhubInput = appDescriptor.getInputStream(inputDescriptor);
+16
+17 //...
+18 }
+{% endhighlight %}
+
+In the code snippet above, we create the input and output streams that can consume and produce from the configured Event Hubs entities.
+
+1. Line 3: A `EventHubsSystemDescriptor` is created with the name "eventhubs". You may set different system descriptors here.
+2. Line 6: Event Hubs messages are consumed as key value pairs. The [serde](../../documentation/versioned/container/serialization.html) is defined for the value of the incoming payload of the Event Hubs' EventData. You may use any of the serdes that samza ships with out of the box or define your own.
+The serde for the key is not set since it will always the String from the EventData [partitionKey](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data._system_properties.getpartitionkey?view=azure-java-stable#com_microsoft_azure_eventhubs__event_data__system_properties_getPartitionKey__).
+3. Line 8-12: An `EventHubsInputDescriptor` is created with the required descriptors to gain access of the Event Hubs entity (`STREAM_ID`, `EVENTHUBS_NAMESPACE`, `EVENTHUBS_ENTITY`, `EVENTHUBS_SAS_KEY_NAME`, `EVENTHUBS_SAS_KEY_TOKEN`).
+These must be set to the credentials of the entities you wish to connect to.
+4. Line 15: creates an `InputStream` with the previously defined `EventHubsInputDescriptor`.
-You can configure your Samza jobs to process data from Azure Event Hubs. To configure Samza to consume from Event Hubs streams:
+Alternatively, you can set these properties in the `.properties` file of the application.
+Note: the keys set in the `.properties` file will override the ones set in code with descriptors.
+Refer to the [Event Hubs configuration reference](../../documentation/versioned/jobs/samza-configurations.html#eventhubs) for the complete list of configurations.
{% highlight jproperties %}
# define an event hub system factory with your identifier. eg: eh-system
@@ -48,7 +84,42 @@ It is required to provide values for YOUR-STREAM-NAMESPACE, YOUR-ENTITY-NAME, YO
## Producing to Event Hubs
-Similarly, you can also configure your Samza job to write to Event Hubs. Follow the same configs defined in the Consuming from Event Hubs section to write to Event Hubs:
+Each [OutgoingMessageEnvelope](https://samza.apache.org/learn/documentation/latest/api/javadocs/org/apache/samza/system/OutgoingMessageEnvelope.html) from Samza is converted into an [EventData](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data) instance whose body is set to the message in the envelope. Additionally, the key and the produce timestamp are set as properties in the EventData before sending it to Event Hubs.
+Similarly, you can also configure your Samza job to write to Event Hubs. Follow the same descriptors defined in the Consuming from Event Hubs section to write to Event Hubs:
+
+{% highlight java %}
+ 1 public void describe(StreamApplicationDescriptor appDescriptor) {
+ 2 // Define your system here
+ 3 EventHubsSystemDescriptor systemDescriptor = new EventHubsSystemDescriptor("eventhubs");
+ 4
+ 5 // Choose your serializer/deserializer for the EventData payload
+ 6 StringSerde serde = new StringSerde();
+ 7
+ 8 // Define the input and output descriptors with respective descriptors
+ 9 EventHubsOutputDescriptor<KV<String, String>> outputDescriptor =
+10 systemDescriptor.getOutputDescriptor(OUTPUT_STREAM_ID, EVENTHUBS_NAMESPACE, EVENTHUBS_OUTPUT_ENTITY, serde)
+11 .withSasKeyName(EVENTHUBS_SAS_KEY_NAME)
+12 .withSasKey(EVENTHUBS_SAS_KEY_TOKEN);
+13
+14 // Define the output streams with descriptors
+15 OutputStream<KV<String, String>> eventhubOutput = appDescriptor.getOutputStream(outputDescriptor);
+16
+17 //...
+18 }
+{% endhighlight %}
+
+In the code snippet above, we create the input and output streams that can consume and produce from the configured Event Hubs entities.
+
+1. Line 3: A `EventHubsSystemDescriptor` is created with the name "eventhubs". You may set different system descriptors here.
+2. Line 6: Event Hubs messages are produced as key value pairs. The [serde](../../documentation/versioned/container/serialization.html) is defined for the value of the payload of the outgoing Event Hubs' EventData. You may use any of the serdes that samza ships with out of the box or define your own.
+The serde for the key is not set since it will always the String from the EventData [partitionKey](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data._system_properties.getpartitionkey?view=azure-java-stable#com_microsoft_azure_eventhubs__event_data__system_properties_getPartitionKey__).
+3. Line 9-12: An `EventHubsOutputDescriptor` is created with the required descriptors to gain access of the Event Hubs entity (`STREAM_ID`, `EVENTHUBS_NAMESPACE`, `EVENTHUBS_ENTITY`, `EVENTHUBS_SAS_KEY_NAME`, `EVENTHUBS_SAS_KEY_TOKEN`).
+These must be set to the credentials of the entities you wish to connect to.
+4. Line 15: creates an `OutputStream` with the previously defined `EventHubsOutputDescriptor`.
+
+Alternatively, you can set these properties in the `.properties` file of the application.
+Note: the keys set in the `.properties` file will override the ones set in code with descriptors.
+Refer to the [Event Hubs configuration reference](../../documentation/versioned/jobs/samza-configurations.html#eventhubs) for the complete list of configurations.
{% highlight jproperties %}
# define an event hub system factory with your identifier. eg: eh-system
@@ -64,15 +135,14 @@ streams.eh-output-stream.eventhubs.sas.keyname=YOUR-SAS-KEY-NAME
streams.eh-output-stream.eventhubs.sas.token=YOUR-SAS-KEY-TOKEN
{% endhighlight %}
-Then you can create and produce a message to Event Hubs in your code as below:
+Then you can consume and produce a message to Event Hubs in your code as below:
{% highlight java %}
-OutgoingMessageEnvelope envelope = new OutgoingMessageEnvelope(new SystemStream("eh-system", "output0"), key, message);
-collector.send(envelope);
+MessageStream<KV<String, String>> eventhubInput = appDescriptor.getInputStream(inputDescriptor);
+OutputStream<KV<String, String>> eventhubOutput = appDescriptor.getOutputStream(outputDescriptor);
+eventhubInput.sendTo(eventhubOutput)
{% endhighlight %}
-Each [OutgoingMessageEnvelope](https://samza.apache.org/learn/documentation/latest/api/javadocs/org/apache/samza/system/OutgoingMessageEnvelope.html) is converted into an [EventData](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data) instance whose body is set to the message in the envelope. Additionally, the key and the produce timestamp are set as properties in the EventData before sending it to Event Hubs.
-
## Advanced configuration
###Producer partitioning
@@ -85,6 +155,13 @@ The partition.method property determines how outgoing messages are partitioned.
3. ROUND\_ROBIN: In this method, outgoing messages are distributed in a round-robin across all partitions. The key and the partition key in the message are ignored.
+##### Using descriptors
+{% highlight java %}
+EventHubsSystemDescriptor systemDescriptor = new EventHubsSystemDescriptor("eventhubs")
+ .withPartitioningMethod(PartitioningMethod.EVENT_HUB_HASHING);
+{% endhighlight %}
+
+##### Using config properties
{% highlight jproperties %}
systems.eh-system.partition.method = EVENT_HUB_HASHING
{% endhighlight %}
@@ -93,6 +170,15 @@ systems.eh-system.partition.method = EVENT_HUB_HASHING
Event Hubs supports the notion of [consumer groups](https://docs.microsoft.com/en-us/azure/event-hubs/event-hubs-features#consumer-groups) which enable multiple applications to have their own view of the event stream. Each partition is exclusively consumed by one consumer in the consumer group. Each event hub stream has a pre-defined consumer group named $Default. You can define your own consumer group for your job by configuring an Event Hubs.consumer.group
+##### Using descriptors
+{% highlight java %}
+EventHubsSystemDescriptor systemDescriptor = new EventHubsSystemDescriptor("eventhubs");
+EventHubsInputDescriptor<KV<String, String>> inputDescriptor =
+ systemDescriptor.getInputDescriptor(INPUT_STREAM_ID, EVENTHUBS_NAMESPACE, EVENTHUBS_INPUT_ENTITY, serde)
+ .withConsumerGroup("my-group");
+{% endhighlight %}
+
+##### Using config properties
{% highlight jproperties %}
streams.eh-input-stream.eventhubs.consumer.group = my-group
{% endhighlight %}
@@ -101,6 +187,17 @@ streams.eh-input-stream.eventhubs.consumer.group = my-group
By default, the messages from Event Hubs are sent and received as byte arrays. You can configure a serializer and deserializer for your message by setting a value for msg.serde for your stream.
+##### Using descriptors
+{% highlight java %}
+JsonSerde inputSerde = new JsonSerde();
+EventHubsInputDescriptor<KV<String, String>> inputDescriptor =
+ systemDescriptor.getInputDescriptor(INPUT_STREAM_ID, EVENTHUBS_NAMESPACE, EVENTHUBS_INPUT_ENTITY, inputSerde);
+JsonSerde outputSerde = new JsonSerde();
+EventHubsOutputDescriptor<KV<String, String>> outputDescriptor =
+ systemDescriptor.getOutputDescriptor(OUTPUT_STREAM_ID, EVENTHUBS_NAMESPACE, EVENTHUBS_OUTPUT_ENTITY, outputSerde);
+{% endhighlight %}
+
+##### Using config properties
{% highlight jproperties %}
streams.input0.samza.msg.serde = json
streams.output0.samza.msg.serde = json
@@ -109,7 +206,13 @@ streams.output0.samza.msg.serde = json
### Consumer buffer size
When the consumer reads a message from event hubs, it appends them to a shared producer-consumer queue corresponding to its partition. This config determines the per-partition queue size. Setting a higher value for this config typically achieves a higher throughput at the expense of increased on-heap memory.
+##### Using descriptors
+{% highlight java %}
+ EventHubsSystemDescriptor systemDescriptor = new EventHubsSystemDescriptor("eventhubs")
+ .withReceiveQueueSize(10);
+{% endhighlight %}
+##### Using config properties
{% highlight jproperties %}
systems.eh-system.eventhubs.receive.queue.size = 10
{% endhighlight %}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/5ba543fa/docs/learn/tutorials/versioned/index.md
----------------------------------------------------------------------
diff --git a/docs/learn/tutorials/versioned/index.md b/docs/learn/tutorials/versioned/index.md
index 97a9fd5..5c22020 100644
--- a/docs/learn/tutorials/versioned/index.md
+++ b/docs/learn/tutorials/versioned/index.md
@@ -32,7 +32,7 @@ title: Tutorials
[Deploying a Samza Job from HDFS](deploy-samza-job-from-hdfs.html)
-[Deoloy Samza to CDH](deploy-samza-to-CDH.html)
+[Deploy Samza to CDH](deploy-samza-to-CDH.html)
[Run Hello-samza in Multi-node YARN](run-in-multi-node-yarn.html)
@@ -44,6 +44,8 @@ title: Tutorials
[Samza SQL User Guide](samza-sql.html)
+[Samza Event Hubs Connectors Example](samza-event-hubs-standalone.html)
+
<!-- TODO a bunch of tutorials
[Log Walkthrough](log-walkthrough.html)
http://git-wip-us.apache.org/repos/asf/samza/blob/5ba543fa/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md
----------------------------------------------------------------------
diff --git a/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md b/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md
new file mode 100644
index 0000000..53dbec3
--- /dev/null
+++ b/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md
@@ -0,0 +1,127 @@
+---
+layout: page
+title: Samza Event Hubs Connectors Example
+---
+<!--
+ 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 [hello-samza](https://github.com/apache/samza-hello-samza) project has an example that uses the Samza high-level API to consume and produce from [Event Hubs](../../documentation/versioned/connectors/eventhubs.html) using the Zookeeper deployment model.
+
+#### Get the Code
+
+Let's get started by cloning the hello-samza project
+
+{% highlight bash %}
+git clone https://git.apache.org/samza-hello-samza.git hello-samza
+cd hello-samza
+git checkout latest
+{% endhighlight %}
+
+The project comes up with numerous examples and for this tutorial, we will pick the Azure Event Hubs demo application.
+
+#### Setting up the Deployment Environment
+
+For our Azure application, we require [ZooKeeper](http://zookeeper.apache.org/). The hello-samza project comes with a script called "grid" to help with the environment setup
+
+{% highlight bash %}
+./bin/grid standalone
+{% endhighlight %}
+
+This command will download, install, and start ZooKeeper and Kafka. It will also check out the latest version of Samza and build it. All package files will be put in a sub-directory called "deploy" inside hello-samza's root folder.
+
+If you get a complaint that JAVA_HOME is not set, then you'll need to set it to the path where Java is installed on your system.
+
+
+#### Configuring the Samza Application
+
+Here are the [Event Hubs descriptors](../../documentation/versioned/connectors/eventhubs.html) you must set before building the project.
+Configure these in the `src/main/java/samza/examples/AzureApplication.java` file.
+
+{% highlight java %}
+ 1 public void describe(StreamApplicationDescriptor appDescriptor) {
+ 2 // Define your system here
+ 3 EventHubsSystemDescriptor systemDescriptor = new EventHubsSystemDescriptor("eventhubs");
+ 4
+ 5 // Choose your serializer/deserializer for the EventData payload
+ 6 StringSerde serde = new StringSerde();
+ 7
+ 8 // Define the input and output descriptors with respective descriptors
+ 9 EventHubsInputDescriptor<KV<String, String>> inputDescriptor =
+10 systemDescriptor.getInputDescriptor(INPUT_STREAM_ID, EVENTHUBS_NAMESPACE, EVENTHUBS_INPUT_ENTITY, serde)
+11 .withSasKeyName(EVENTHUBS_SAS_KEY_NAME)
+12 .withSasKey(EVENTHUBS_SAS_KEY_TOKEN);
+13
+14 EventHubsOutputDescriptor<KV<String, String>> outputDescriptor =
+15 systemDescriptor.getOutputDescriptor(OUTPUT_STREAM_ID, EVENTHUBS_NAMESPACE, EVENTHUBS_OUTPUT_ENTITY, serde)
+16 .withSasKeyName(EVENTHUBS_SAS_KEY_NAME)
+17 .withSasKey(EVENTHUBS_SAS_KEY_TOKEN);
+18
+19 // Define the input and output streams with descriptors
+20 MessageStream<KV<String, String>> eventhubInput = appDescriptor.getInputStream(inputDescriptor);
+21 OutputStream<KV<String, String>> eventhubOutput = appDescriptor.getOutputStream(outputDescriptor);
+22
+23 //...
+24 }
+{% endhighlight %}
+
+In the code snippet above, we create the input and output streams that can consume and produce from the configured Event Hubs entities.
+
+1. Line 3: A `EventHubsSystemDescriptor` is created with the name "eventhubs". You may set different system descriptors here.
+2. Line 6: Event Hubs messages are consumed as key value pairs. The [serde](../../documentation/versioned/container/serialization.html) is defined for the value of the payload of the Event Hubs' EventData. You may use any of the serdes that samza ships with out of the box or define your own.
+The serde for the key is not set since it will always the String from the EventData [partitionKey](https://docs.microsoft.com/en-us/java/api/com.microsoft.azure.eventhubs._event_data._system_properties.getpartitionkey?view=azure-java-stable#com_microsoft_azure_eventhubs__event_data__system_properties_getPartitionKey__).
+3. Line 9-17: An `EventHubsInputDescriptor` and an `EventHubsOutputDescriptor` are created with the required descriptors to gain access of the Event Hubs entity (`STREAM_ID`, `EVENTHUBS_NAMESPACE`, `EVENTHUBS_ENTITY`, `EVENTHUBS_SAS_KEY_NAME`, `EVENTHUBS_SAS_KEY_TOKEN`).
+These must be set to the credentials of the entities you wish to connect to.
+4. Line 10-21: creates an `InputStream` and `OutputStream` with the previously defined `EventHubsInputDescriptor` and `EventHubsOutputDescriptor`, respectively.
+
+Alternatively, you can set these properties in the `src/main/config/azure-application-local-runner.properties` file.
+Note: the keys set in the `.properties` file will override the ones set in code with descriptors.
+Refer to the [Event Hubs configuration reference](../../documentation/versioned/jobs/samza-configurations.html#eventhubs) for the complete list of configurations.
+
+#### Building the Hello Samza Project
+
+With the environment setup complete, let's move on to building the hello-samza project. Execute the following command:
+
+{% highlight bash %}
+./bin/deploy.sh
+{% endhighlight %}
+
+We are now all set to run the application locally.
+
+#### Running the Azure application
+
+In order to run the application, we will use the *run-azure-application* script.
+
+{% highlight bash %}
+./deploy/samza/bin/run-event-hubs-zk-application.sh
+{% endhighlight %}
+
+The above command executes the helper script which invokes the *AzureZKLocalApplication* main class, which starts the *AzureApplication*. This application prints out the messages from the input stream to `stdout` and send them the output stream.
+
+The messages consumed should be printed in the following format:
+{% highlight bash %}
+Sending:
+Received Key: <KEY>
+Received Message: <VALUE>
+{% endhighlight %}
+
+#### Shutdown
+
+This application can be shutdown by terminating the *run-azure-application* script.
+We can use the *grid* script to tear down the local environment ([Kafka](http://kafka.apache.org/) and [Zookeeper](http://zookeeper.apache.org/)).
+
+{% highlight bash %}
+./bin/grid stop all
+{% endhighlight %}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/5ba543fa/samza-azure/src/test/java/org/apache/samza/system/eventhub/consumer/ITestEventHubSystemConsumer.java
----------------------------------------------------------------------
diff --git a/samza-azure/src/test/java/org/apache/samza/system/eventhub/consumer/ITestEventHubSystemConsumer.java b/samza-azure/src/test/java/org/apache/samza/system/eventhub/consumer/ITestEventHubSystemConsumer.java
index cfd8217..e36635e 100644
--- a/samza-azure/src/test/java/org/apache/samza/system/eventhub/consumer/ITestEventHubSystemConsumer.java
+++ b/samza-azure/src/test/java/org/apache/samza/system/eventhub/consumer/ITestEventHubSystemConsumer.java
@@ -66,8 +66,8 @@ public class ITestEventHubSystemConsumer {
if (numEvents > 0) {
EventHubIncomingMessageEnvelope eventData = (EventHubIncomingMessageEnvelope) result.get(0);
System.out.println("System properties: " + eventData.getEventData().getSystemProperties());
+ System.out.println("Key: " + new String((byte[]) eventData.getKey()));
System.out.println("Message: " + new String((byte[]) eventData.getMessage()));
- break;
}
System.out.println("Retries left: " + numRetries);
}
http://git-wip-us.apache.org/repos/asf/samza/blob/5ba543fa/samza-azure/src/test/java/org/apache/samza/system/eventhub/producer/ITestEventHubSystemProducer.java
----------------------------------------------------------------------
diff --git a/samza-azure/src/test/java/org/apache/samza/system/eventhub/producer/ITestEventHubSystemProducer.java b/samza-azure/src/test/java/org/apache/samza/system/eventhub/producer/ITestEventHubSystemProducer.java
index 63f6daa..2c35a5c 100644
--- a/samza-azure/src/test/java/org/apache/samza/system/eventhub/producer/ITestEventHubSystemProducer.java
+++ b/samza-azure/src/test/java/org/apache/samza/system/eventhub/producer/ITestEventHubSystemProducer.java
@@ -76,7 +76,7 @@ public class ITestEventHubSystemProducer {
systemProducer.send(STREAM_NAME1, createMessageEnvelope(STREAM_NAME1));
try {
- systemProducer.send(STREAM_NAME2, createMessageEnvelope(STREAM_NAME1));
+ systemProducer.send("unregistered_stream", createMessageEnvelope("unregistered_stream"));
Assert.fail("Sending event to destination that is not registered should throw exception");
} catch (SamzaException e) {
}