How to send compressed Json to azure event hub and read in fabric pyspark streaming ?

Kumar, Sudarshan 0

I am sending compressed data into Event Hub to overcome 1 MB hard limit in Azure Event Hub. I also have to read this in Py-spark and update delta table.

The compressed data send to Event hub is coming as null in Py-spark stream. How to read it?

This is how I am reading from Event Hub

df_stream_body = df_stream.select(F.from_json(F.col("body").cast("string"), message_schema).alias("Payload"))

This is how I am sending data to Event Hub

` async with producer:

        # Create a batch.
        event_data_batch = await producer.create_batch()

        
        # Add events to the batch.
        body = '{"id": "200", "firstName": "Sudarshan70","middleName": "Kumar2","lastName": "Thakur2"}'  

        
        # Compress the JSON string using the gzip algorithm.
        compressed_body = gzip.compress(body.encode('utf-8'))

        event_data_batch.add(EventData(compressed_body))
        # Send the batch of events to the event hub.
        await producer.send_batch(event_data_batch)`
```when I compress and send data to event hub and read in spark i get null in all filed .

PRADEEPCHEEKATLA-MSFT 79,141 Reputation points Microsoft Employee

2024-05-01T08:53:54.4566667+00:00
@Kumar, Sudarshan - Thanks for the question and using MS Q&A platform.

It seems that you are still facing issues with reading the compressed data in PySpark.

Based on the code you have shared, it seems that you are compressing the JSON string using the gzip algorithm before sending it to Event Hub. However, you are not setting the compression property in the EventData constructor.

To read compressed data from Event Hub in PySpark, you need to set the compression property in the EventData constructor to "gzip". This will tell Event Hub that the data is compressed using the gzip algorithm.

Here is an example of how you can set the compression property in the EventData constructor:

body = '{"id": "200", "firstName": "Sudarshan70","middleName": "Kumar2","lastName": "Thakur2"}' compressed_body = gzip.compress(body.encode('utf-8')) event_data = EventData(compressed_body, compression="gzip") await producer.send(event_data)

Once you have set the compression property and sent the compressed data to Event Hub, you can read it in PySpark using the following code:

df_stream_body = df_stream.select(F.from_json(F.decompress(F.col("body")).cast("string"), message_schema).alias("Payload"))

Here, we are using the decompress function from the pyspark.sql.functions module to decompress the data before parsing it as JSON using the from_json function.

I hope this helps! Let me know if you have any further questions.
Kumar, Sudarshan 0 Reputation points

2024-05-01T08:59:19.79+00:00
this is how I am reading from Event hub using stream .Is this correct way ?

df_stream = spark.readStream.format("eventhubs")\ .options(**ehConf)\ .option("compression", "gzip") \ .load()

and while sending when i used compression properties i got below error

"errorMessage": "__init__() got an unexpected keyword argument 'compression'",
PRADEEPCHEEKATLA-MSFT 79,141 Reputation points Microsoft Employee

2024-05-02T03:24:23.97+00:00
@Kumar, Sudarshan - The code you have shared for reading from Event Hub using PySpark looks correct. You are using the eventhubs format and setting the compression option to "gzip", which should allow you to read compressed data from Event Hub.

Regarding the error you are seeing when setting the compression property in the EventData constructor, it seems that the compression property is not supported in the version of the Azure Event Hubs SDK that you are using. The compression property was added in version 5.2.0 of the Azure Event Hubs SDK for Python.

To resolve this issue, you can try upgrading the Azure Event Hubs SDK to version 5.2.0 or later. You can do this by running the following command:

pip install azure-eventhub==5.2.0

Once you have upgraded the SDK, you should be able to use the compression property in the EventData constructor to send compressed data to Event Hub.

I hope this helps! Let me know if you have any further questions.
Kumar, Sudarshan 0 Reputation points

2024-05-02T09:50:56.81+00:00

I am still getting error in latest version
PRADEEPCHEEKATLA-MSFT 79,141 Reputation points Microsoft Employee

2024-05-06T05:53:21.4566667+00:00

@Kumar, Sudarshan - I agree that this issue looks strange and I wasn't able to reproduce this issue. If you have a support plan could you please file a support ticket for deeper investigation and do share the SR# with us?

Share via

How to send compressed Json to azure event hub and read in fabric pyspark streaming ?