Processing event hub data using python

0
votes

I am using azure event hub python SDK to send to and receive messages from event hub following this link.https://github.com/Azure/azure-event-hubs-python/tree/develop. I can successfully send and receive messages. But how do i parse the messages and retrieve the data from the event data object. Please find the code below.

import os
import sys
#import logging
from azure.eventhub import EventHubClient, Receiver, Offset

ADDRESS = 'sb://####.servicebus.windows.net/#####'
USER = '##########'
KEY = '##################################'
CONSUMER_GROUP = "$default"
OFFSET = Offset("-1")
PARTITION = "1"


total = 0
last_sn = -1
last_offset = "-1"

try:
  if not ADDRESS:
      raise ValueError("No EventHubs URL supplied.")
  client = EventHubClient(ADDRESS, debug=False, username=USER, password=KEY)
  receiver = client.add_receiver(CONSUMER_GROUP, PARTITION, prefetch=5000, 
  offset=OFFSET)
  client.run()
  try:
      batched_events = receiver.receive(timeout=20)
  except:
      raise
  finally:
      client.stop()
  for event_data in batched_events:
      last_offset = event_data.offset.value
      last_sn = event_data.sequence_number
      total += 1
      print("Partition {}, Received {}, sn={} offset={}".format(
         PARTITION,
         total,
         last_sn,
         last_offset))

except KeyboardInterrupt:
   pass

if i try to view the event_data received i can see the below message. event_data <azure.eventhub.common.EventData at 0xd4f1358> event_data.message

<uamqp.message.Message at 0xd4f1240>

Any help on the above on how to parse this message to extract the data

2
pythonazureazure-eventhubazure-eventhub-capture
Try event_data.body Reference: github.com/Azure/azure-event-hubs-python/blob/master/azure/… – Shiva
It gives me this <generator object data at 0x000000000D547DB0>. how to get the str data from this. Any suggestion? – sruthi
It is a generator object, it will return a value on every next call i.e. next(event_data.body). If you want all the values at once, do list(event_data.body). – Shiva
@Shiva You could add it as an answer, if it solved the issue. – Tom Sun - MSFT
Thanks @TomSun, posted an elaborate answer. – Shiva

2 Answers

3
votes

As of 1.1.0, there are new utility methods to extract the actual data of the message:

So, what used to be 

import json
event_obj = json.loads(next(event_data.body).decode('UTF-8'))

Is now:

event_obj = event_data.body_as_json()
1
votes

For people using the Event Hub version 5.2.0 -- latest as of today (GitHub, Reference Docs), it's the same as the 1.1.0 version, i.e. use body_as_str() or body_as_json(). But the client has changed -- there's an EventHubProducerClient and an EventHubConsumerClient in the new version. To print the body of an event received:

from azure.eventhub import EventHubConsumerClient

connection_str = '<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>'
consumer_group = '<< CONSUMER GROUP >>'
eventhub_name = '<< NAME OF THE EVENT HUB >>'

client = EventHubConsumerClient.from_connection_string(
        connection_str, consumer_group, eventhub_name=eventhub_name
    )

def on_event_batch(partition_context, events):
    partition_context.update_checkpoint()
    for e in events:
        print(e.body_as_str())

with client:
    client.receive_batch(
        on_event_batch=on_event_batch,
        starting_position="-1",  # "-1" is from the beginning of the partition.
    )