Victor Leung

Azure EventHub, LogicApp and DataVerse

Victor Leung's photo
Victor Leung
·

3 min read

Kafka messages can be exported and imported to Microsoft Cloud for Financial Services (FSI). It provides different components, such as a unified customer profile to manage customer data, and it could be used to store personally identifiable information (PII) data. The connection could be made; where data flow from Kafka to Azure EventHub, it uses LogicApp to further sync the data to DataVerse, which FSI could consume as shown in the diagram below:

Follow the steps below to setup this connection:

  1. Send event to Azure EventHub.

For example, the python script below could use to send three simple event messages to Azure EventHub.

import time
import os
from azure.eventhub import EventHubProducerClient, EventData
from azure.eventhub.exceptions import EventHubError
import json

EVENTHUB_NAME = "REPLACE_WITH_EVENTHUB_NAME"
CONNECTION_STR = "Endpoint=sb://REPLEASE_WITH_CONNECTION_STRING.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=REPLACE_WITH_SHARED_ACCESS_KEY"

body = json.dumps({
    "id": "something"
})

def send_event_data_batch(producer, i):
    # Without specifying partition_id or partition_key
    # the events will be distributed to available partitions via round-robin.
    event_data_batch = producer.create_batch()
    event_data_batch.add(EventData(body))
    producer.send_batch(event_data_batch)

producer = EventHubProducerClient.from_connection_string(
    conn_str=CONNECTION_STR,
    eventhub_name=EVENTHUB_NAME
)

start_time = time.time()
with producer:
    for i in range(3):
        send_event_data_batch(producer, i)

print("Send messages in {} seconds.".format(time.time() - start_time))

Replace the placeholder of the event hub name and connection string. If the message was sent successfully, you would see the message, like Send messages in 1.730254888534546 seconds.

If you got an error “Authentication Put-Token failed. Retries exhausted“, double-check the placeholder values and see if it’s correct.

  1. Azure Eventhub to LogicApp

Navigate to the Azure portal, search for LogicApp and create a new one. It can be used as automated workflows, where EventHub events as the trigger and DataVerse as the output. Fill in the webform to create a Logic App. You can select plan type as Consumption, which is best for entry-level development purposes.

Once the LogicApp is created, navigate to Development Tools for the Logic app designer. We would make three steps in the process.

2.1 EventHub Trigger In the first step of the workflow, we would connect to EventHub as the trigger. For development purposes, shorten the check items time to 3 seconds. Follow the UI to select and connect your existing EventHub name with an example shown below:

2.2. Initialize variables Next step, we would parse the events hubs message. The example message sent above is:

{
    "id": "something"
}

Where we want to get the value using the key “id”, this could be done via expression:

json(decodeBase64(triggerBody()['ContentData']))['id']

2.3 Add a row to Data Verse Finally, the third step is to use the database connector. We can link to the corresponding Dataverse Tables to add a new row. If the table doesn’t exist yet, go to make.powerapps.com , select Dataverse, Tables to create a Table. The corresponding field can be filled in using step 2 initialized variables.

Once it’s done, click Save of the workflow.

3.Dataverse

Dataverse is a database for storing data with Tables. If the LogicApp is triggered successfully when a new event is added, then you should see the new row is added to the DataVerse table

Finally, once all the Vault data are synced to Azure FSI, the component can consume it, such as the Unified customer profile to manage customer data.

Navigate to Microsoft Cloud Solution Center https://solutions.microsoft.com/ with the account credentials to select the component.

To launch the Dynamic 365 sandbox, navigate the Solution centre and click the “Launch” button.

There are populated sample data for the Unified customer profile app like:

Let me know if you got any questions setting this up. Cheers.

AzureCloudkafkadata