-
Notifications
You must be signed in to change notification settings - Fork 2.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[SchemaRegistry] Samples for EH integration (#13884)
* add sample for EH integration * add samples to readme and tweak the code * add descriptions * mention SR and serializer in EH * small tweak
- Loading branch information
1 parent
ad2142b
commit 243212a
Showing
6 changed files
with
257 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
64 changes: 64 additions & 0 deletions
64
...chemaregistry/azure-schemaregistry-avroserializer/samples/eventhub_receive_integration.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,64 @@ | ||
| #!/usr/bin/env python | ||
|
|
||
| # -------------------------------------------------------------------------------------------- | ||
| # Copyright (c) Microsoft Corporation. All rights reserved. | ||
| # Licensed under the MIT License. See License.txt in the project root for license information. | ||
| # -------------------------------------------------------------------------------------------- | ||
|
|
||
| """ | ||
| Examples to show receiving events from EventHub with SchemaRegistryAvroSerializer integrated for data deserialization. | ||
| """ | ||
|
|
||
| # pylint: disable=C0111 | ||
| import os | ||
| from azure.eventhub import EventHubConsumerClient | ||
| from azure.identity import DefaultAzureCredential | ||
| from azure.schemaregistry import SchemaRegistryClient | ||
| from azure.schemaregistry.serializer.avroserializer import SchemaRegistryAvroSerializer | ||
|
|
||
| EVENTHUB_CONNECTION_STR = os.environ['EVENT_HUB_CONN_STR'] | ||
| EVENTHUB_NAME = os.environ['EVENT_HUB_NAME'] | ||
|
|
||
| SCHEMA_REGISTRY_ENDPOINT = os.environ['SCHEMA_REGISTRY_ENDPOINT'] | ||
| SCHEMA_GROUP = os.environ['SCHEMA_REGISTRY_GROUP'] | ||
|
|
||
|
|
||
| def on_event(partition_context, event): | ||
| print("Received event from partition: {}.".format(partition_context.partition_id)) | ||
|
|
||
| bytes_payload = b"".join(b for b in event.body) | ||
| print('The received bytes of the EventData is {}.'.format(bytes_payload)) | ||
|
|
||
| # Use the deserialize method to convert bytes to dict object. | ||
| # The deserialize method would extract the schema id from the payload, and automatically retrieve the Avro Schema | ||
| # from the Schema Registry Service. The schema would be cached locally for future usage. | ||
| deserialized_data = avro_serializer.deserialize(bytes_payload) | ||
| print('The dict data after deserialization is {}'.format(deserialized_data)) | ||
|
|
||
|
|
||
| # create an EventHubConsumerClient instance | ||
| eventhub_consumer = EventHubConsumerClient.from_connection_string( | ||
| conn_str=EVENTHUB_CONNECTION_STR, | ||
| consumer_group='$Default', | ||
| eventhub_name=EVENTHUB_NAME, | ||
| ) | ||
|
|
||
|
|
||
| # create a SchemaRegistryAvroSerializer instance | ||
| avro_serializer = SchemaRegistryAvroSerializer( | ||
| schema_registry=SchemaRegistryClient( | ||
| endpoint=SCHEMA_REGISTRY_ENDPOINT, | ||
| credential=DefaultAzureCredential() | ||
| ), | ||
| schema_group=SCHEMA_GROUP | ||
| ) | ||
|
|
||
|
|
||
| try: | ||
| with eventhub_consumer, avro_serializer: | ||
| eventhub_consumer.receive( | ||
| on_event=on_event, | ||
| starting_position="-1", # "-1" is from the beginning of the partition. | ||
| ) | ||
| except KeyboardInterrupt: | ||
| print('Stopped receiving.') |
72 changes: 72 additions & 0 deletions
72
sdk/schemaregistry/azure-schemaregistry-avroserializer/samples/eventhub_send_integration.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| #!/usr/bin/env python | ||
|
|
||
| # -------------------------------------------------------------------------------------------- | ||
| # Copyright (c) Microsoft Corporation. All rights reserved. | ||
| # Licensed under the MIT License. See License.txt in the project root for license information. | ||
| # -------------------------------------------------------------------------------------------- | ||
|
|
||
| """ | ||
| Examples to show sending event to EventHub with SchemaRegistryAvroSerializer integrated for data serialization. | ||
| """ | ||
|
|
||
| # pylint: disable=C0111 | ||
|
|
||
| import os | ||
| from azure.eventhub import EventHubProducerClient, EventData | ||
| from azure.identity import DefaultAzureCredential | ||
| from azure.schemaregistry import SchemaRegistryClient | ||
| from azure.schemaregistry.serializer.avroserializer import SchemaRegistryAvroSerializer | ||
|
|
||
| EVENTHUB_CONNECTION_STR = os.environ['EVENT_HUB_CONN_STR'] | ||
| EVENTHUB_NAME = os.environ['EVENT_HUB_NAME'] | ||
|
|
||
| SCHEMA_REGISTRY_ENDPOINT = os.environ['SCHEMA_REGISTRY_ENDPOINT'] | ||
| SCHEMA_GROUP = os.environ['SCHEMA_REGISTRY_GROUP'] | ||
|
|
||
| SCHEMA_STRING = """ | ||
| {"namespace": "example.avro", | ||
| "type": "record", | ||
| "name": "User", | ||
| "fields": [ | ||
| {"name": "name", "type": "string"}, | ||
| {"name": "favorite_number", "type": ["int", "null"]}, | ||
| {"name": "favorite_color", "type": ["string", "null"]} | ||
| ] | ||
| }""" | ||
|
|
||
|
|
||
| def send_event_data_batch(producer, serializer): | ||
| event_data_batch = producer.create_batch() | ||
|
|
||
| dict_data = {"name": "Bob", "favorite_number": 7, "favorite_color": "red"} | ||
| # Use the serialize method to convert dict object to bytes with the given avro schema. | ||
| # The serialize method would automatically register the schema into the Schema Registry Service and | ||
| # schema would be cached locally for future usage. | ||
| payload_bytes = serializer.serialize(data=dict_data, schema=SCHEMA_STRING) | ||
| print('The bytes of serialized dict data is {}.'.format(payload_bytes)) | ||
|
|
||
| event_data = EventData(body=payload_bytes) # pass the bytes data to the body of an EventData | ||
| event_data_batch.add(event_data) | ||
| producer.send_batch(event_data_batch) | ||
| print('Send is done.') | ||
|
|
||
|
|
||
| # create an EventHubProducerClient instance | ||
| eventhub_producer = EventHubProducerClient.from_connection_string( | ||
| conn_str=EVENTHUB_CONNECTION_STR, | ||
| eventhub_name=EVENTHUB_NAME | ||
| ) | ||
|
|
||
|
|
||
| # create a SchemaRegistryAvroSerializer instance | ||
| avro_serializer = SchemaRegistryAvroSerializer( | ||
| schema_registry=SchemaRegistryClient( | ||
| endpoint=SCHEMA_REGISTRY_ENDPOINT, | ||
| credential=DefaultAzureCredential() | ||
| ), | ||
| schema_group=SCHEMA_GROUP | ||
| ) | ||
|
|
||
|
|
||
| with eventhub_producer, avro_serializer: | ||
| send_event_data_batch(eventhub_producer, avro_serializer) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters