How to deserialize an IDictionary Object load into an Azure EventHub

Question

How to deserialize an IDictionary Object load into an Azure EventHub

BEPV 0

I am trying to deserialize data that is coming from Azure Event Hub using Kafka integration, the data is loaded into the EventHub using .net app and the message is an IDictionary object as shown below, after reading from the EventHub the data that is loaded into the dataframe is a binary value and I tried to deserialice in python using an explicit cast to string and also using hex function but I am still not able to deserialize the data correctly, could you please guide me on how to deserialize it.

### Read message using Kafka integration

sasl_config = f'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="{client_id}" clientSecret="{client_secret}" scope="https://{event_hubs_server}/.default" ssl.protocol="SSL";'
kafka_options = {
    # Port 9093 is the EventHubs Kafka port
    "kafka.bootstrap.servers": f"{event_hubs_server}:9093",
    "kafka.sasl.jaas.config": sasl_config,
    "kafka.sasl.oauthbearer.token.endpoint.url": f"https://login.microsoft.com/{tenant_id}/oauth2/v2.0/token",
    "subscribe": event_hubs_topic,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.mechanism": "OAUTHBEARER",
    "startingOffsets":"earliest",
    "kafka.sasl.login.callback.handler.class": "org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler",
    "maxOffsetsPerTrigger": 1000
}
df = spark.readStream.format("kafka").options(**kafka_options).load()

### Save message into EventHub

Dictionary<string, object> metadata = new Dictionary<string, object>
{
    ["Id"] = Id,
    ["Document"] = Document,   
    ["IsTest"] = false
};

Dictionary<byte[], IDictionary<string, object>> messages = new Dictionary<byte[], IDictionary<string, object>>() { { guid.ToByteArray(), metadata } };
await EventHubProducer.SendBatchMessageAsync(messages, Id).ConfigureAwait(false);

AnnuKumari-MSFT 34,556 Reputation points Microsoft Employee Moderator

2024-04-16T10:44:25.7066667+00:00
Hi BEPV ,

Welcome to Microsoft Q&A platform and thanks for posting your query here.

As per my understanding you are trying to deserialize data that is coming from Azure Event Hub .

It seems like the data you are receiving from the Kafka integration is in binary format. To deserialize the data, you need to first convert the binary data to a string and then deserialize it into a dictionary object.

Below is the reference code snippet for the same:

import json from pyspark.sql.functions import from_json from pyspark.sql.types import StructType, StructField, StringType, IntegerType # Define the schema for the incoming data schema = StructType([ StructField("Id", StringType()), StructField("Document", StringType()), StructField("IsTest", IntegerType()) ]) # Convert the binary data to a string df = df.selectExpr("CAST(value AS STRING)") # Deserialize the data into a dictionary object df = df.select(from_json(df.value, schema).alias("data")) # Extract the dictionary object from the dataframe df = df.select("data.*") # Now you can access the dictionary object as a regular Python dictionary for row in df.rdd.collect(): print(row["Id"], row["Document"], row["IsTest"])

Additionally, kindly check the below documentation if it can be of help EventHub.DeserializeFromDictionary(IDictionary) Method to deserializes a IDictionary into an instance of EventHub.

Hope it helps. Thankyou.
BEPV 0 Reputation points

2024-04-17T03:26:04.6266667+00:00

That did not work, it shows all the fields as 'null', the documentation does not provide info on how to deserialize it in PySpark.
BEPV 0 Reputation points

2024-04-17T03:27:45.7+00:00

is the IDictionary serialized as json?

Your answer

AnnuKumari-MSFT 34,556 Reputation points Microsoft Employee Moderator

2024-04-16T10:44:25.7066667+00:00

Hi BEPV ,

Welcome to Microsoft Q&A platform and thanks for posting your query here.

As per my understanding you are trying to deserialize data that is coming from Azure Event Hub .

It seems like the data you are receiving from the Kafka integration is in binary format. To deserialize the data, you need to first convert the binary data to a string and then deserialize it into a dictionary object.

Below is the reference code snippet for the same:

import json from pyspark.sql.functions import from_json from pyspark.sql.types import StructType, StructField, StringType, IntegerType # Define the schema for the incoming data schema = StructType([ StructField("Id", StringType()), StructField("Document", StringType()), StructField("IsTest", IntegerType()) ]) # Convert the binary data to a string df = df.selectExpr("CAST(value AS STRING)") # Deserialize the data into a dictionary object df = df.select(from_json(df.value, schema).alias("data")) # Extract the dictionary object from the dataframe df = df.select("data.*") # Now you can access the dictionary object as a regular Python dictionary for row in df.rdd.collect(): print(row["Id"], row["Document"], row["IsTest"])

Additionally, kindly check the below documentation if it can be of help EventHub.DeserializeFromDictionary(IDictionary) Method to deserializes a IDictionary into an instance of EventHub.

Hope it helps. Thankyou.
BEPV 0 Reputation points

2024-04-17T03:26:04.6266667+00:00

That did not work, it shows all the fields as 'null', the documentation does not provide info on how to deserialize it in PySpark.
BEPV 0 Reputation points

2024-04-17T03:27:45.7+00:00

is the IDictionary serialized as json?

Share via

How to deserialize an IDictionary Object load into an Azure EventHub

Your answer