CODEWITHSUNDEEP
pythonazureazure-eventhub
J.C Guzman
J.C Guzman

Reputation: 1334

consume events in python azure event hub

I am receiving JSON data in event hub.

Once a day I want to read this data from event hub and store it in a database. In order to read the data from event hub I am following this documentation: https://learn.microsoft.com/en-us/python/api/overview/azure/eventhub-readme?view=azure-python

I am able to print all the events that I have in my event hub, but I don't know how to get these events and return a pandas dataframe outside of this function.

I have tried this:


def on_event_batch(partition_context, events):
    final_dataframe = pd.DataFrame()
    print("Received event from partition {}".format(partition_context.partition_id))
    for event in events:
        
        body = json.loads(next(event.body).decode('UTF-8'))
        event_df  = pd.DataFrame(body,index = [0])
        final_dataframe = pd.concat([final_dataframe,event_df],ignore_index= True)
    partition_context.update_checkpoint()
    client.close()
    print(final_dataframe)
    return final_dataframe
    
with client:
    final_dataframe = client.receive_batch(
        on_event_batch=on_event_batch, 
        starting_position="-1",  # "-1" is from the beginning of the partition.
    )
    
    # receive events from specified partition:
    # client.receive_batch(on_event_batch=on_event_batch, partition_id='0')

but it is not working.

Upvotes: 1

Views: 3260

Answers (1)

Satya V
Satya V

Reputation: 4174

The client.receive_batch(on_event_batch=on_event_batch, partition_id='0') has return type of None. I am not sure whether you will be able to achieve this by doing a return at the callback function.

However, the easier approach I think would be like below

from azure.eventhub import EventHubConsumerClient
import pandas as pd
import json



def get_messages() :
    connection_str = '<YOUR CONNECTION STRING>'
    consumer_group = '<YOUR CONSUMER GROUP>'
    eventhub_name = '<YOUR EVENT HUB>'
    client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group, eventhub_name=eventhub_name)

    final_df = pd.DataFrame()
    def on_event_batch(partition_context, events):
        print("Received event from partition {}".format(partition_context.partition_id))
        print(len(events))
        #Checking whether there is any event returned as we have set max_wait_time
        if(len(events) == 0):
        #closing the client if there is no event triggered.
            client.close()
            
        else:
        
            for event in events:
                #Event.body operation
                body=event.body
                event_df  = pd.DataFrame(body,index = [0])
                nonlocal final_df
                final_df = pd.concat([final_df,event_df],ignore_index= True)
                partition_context.update_checkpoint()

    with client:
        client.receive_batch(
            on_event_batch=on_event_batch, 
            starting_position="-1",max_wait_time = 5,max_batch_size=2  # "-1" is from the beginning of the partition. 
            #Max_wait_time - no activitiy for that much - call back function is called with No events.
        )
    return final_df


df = get_messages()
df.head()

The above code will actually set the values to the dataframe df after gracefully exiting.

enter image description here

Upvotes: 2

Related Questions