Connect to eventhub using Managed service identity (MSI) from Azure synapse analytics using pyspark spark streaming

Sunil Yadav 0

I am trying to write data to Azure Event Hubs from an Azure Synapse Analytics notebook in PySpark using Managed Service Identity (MSI). Could you please provide step-by-step guidance on how to achieve this, along with the necessary code?

Please note, I have successfully connected using the SAS key previously.

phemanth 13,465 Microsoft Vendor

@Sunil Yadav

Thanks for posting your question in the Microsoft Q&A forum.

Here’s a step-by-step guide to connect to Azure Event Hubs using Managed Service Identity (MSI) from Azure Synapse Analytics using PySpark and Spark Streaming:

Install Required Libraries: Ensure you have the necessary libraries installed in your Synapse notebook. You can install them using the following commands

%%configure -f
{
    "conf": {
        "spark.jars.packages": "com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.18,com.azure:azure-identity:1.4.1"
    }
}

Set Up Event Hubs Configuration: Define the necessary configurations for connecting to your Event Hub using MSI.

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from azure.identity import DefaultAzureCredential
from azure.eventhub import EventHubProducerClient, EventData
# Initialize Spark session
spark = SparkSession.builder.appName("EventHubMSI").getOrCreate()
# Define Event Hub parameters
event_hub_namespace = "<Your Event Hubs Namespace>"
event_hub_name = "<Your Event Hub Name>"
# Create a token credential using Managed Identity
credential = DefaultAzureCredential()
# Build the Event Hubs connection string
connection_string = f"Endpoint=sb://{event_hub_namespace}.servicebus.windows.net/;EntityPath={event_hub_name}"
# Create Event Hub producer client
producer = EventHubProducerClient.from_connection_string(
    conn_str=connection_string,
    credential=credential
)

Write Data to Event Hubs: Use Spark Streaming to write data to Event Hubs

# Sample data to write to Event Hubs
data = [("Hello, Event Hubs!")]
# Create DataFrame
df = spark.createDataFrame(data, StringType())
# Write the data to Event Hubs
query = df.selectExpr("CAST(value AS STRING) AS body") \
    .writeStream \
    .format("eventhubs") \
    .option("eventhubs.connectionString", connection_string) \
    .option("checkpointLocation", "/mnt/checkpoints") \
    .start()
query.awaitTermination()

Verify and Monitor: Ensure that your data is being written to the Event Hub by monitoring the Event Hub metrics in the Azure portal.

Hope this helps. Do let us know if you any further queries.

phemanth 13,465 Reputation points Microsoft Vendor

2024-09-27T12:07:02.45+00:00

@Sunil Yadav We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.
Sunil Yadav 0 Reputation points

2024-09-27T14:04:08.7333333+00:00

@phemanth we have to read data from event-hub by using spark-streaming Like spark.readStream.option
phemanth 13,465 Reputation points Microsoft Vendor

2024-10-01T06:51:43.3966667+00:00

@Sunil Yadav just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.
Ashish Kumar 0 Reputation points Microsoft Employee

2025-01-02T16:06:42.1466667+00:00

Hi @phemanth ,

I was trying to do what you mentioned above but i am getting the following error .

ValueError: At least one of the SharedAccessKey or SharedAccessSignature must be present.

phemanth 13,465 Microsoft Vendor

@Ashish Kumar

The error is indicating that the connection string requires either a SharedAccessKey or SharedAccessSignature. When using Managed Service Identity (MSI), you don't need to include these in the connection string. Instead, you should use the EventHubProducerClient with the credential parameter directly.

Let's adjust the code to use MSI properly:

Step 1: Install Required Libraries

Ensure you have the necessary libraries installed in your Synapse notebook:

%%configure -f
{
    "conf": {
        "spark.jars.packages": "com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.18,com.azure:azure-identity:1.4.1"
    }
}

Step 2: Set Up Event Hubs Configuration

Define the necessary configurations for connecting to your Event Hub using MSI.

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from azure.identity import DefaultAzureCredential
from azure.eventhub import EventHubProducerClient, EventData

# Initialize Spark session
spark = SparkSession.builder.appName("EventHubMSI").getOrCreate()

# Define Event Hub parameters
event_hub_namespace = "<Your Event Hubs Namespace>"
event_hub_name = "<Your Event Hub Name>"

# Create a token credential using Managed Identity
credential = DefaultAzureCredential()

# Create Event Hub producer client
producer = EventHubProducerClient(
    fully_qualified_namespace=f"{event_hub_namespace}.servicebus.windows.net",
    eventhub_name=event_hub_name,
    credential=credential
)

Step 3: Write Data to Event Hubs

Use Spark Streaming to write data to Event Hubs.

# Sample data to write to Event Hubs
data = [("Hello, Event Hubs!")]

# Create DataFrame
df = spark.createDataFrame(data, StringType())

# Write the data to Event Hubs
query = df.selectExpr("CAST(value AS STRING) AS body") \
    .writeStream \
    .format("eventhubs") \
    .option("eventhubs.connectionString", connection_string) \
    .option("checkpointLocation", "/mnt/checkpoints") \
    .start()

query.awaitTermination()

Step 4: Verify and Monitor

Ensure that your data is being written to the Event Hub by monitoring the Event Hub metrics in the Azure portal.

1 answer

phemanth 13,465 Reputation points Microsoft Vendor

2024-09-30T10:34:51.07+00:00
@Sunil Yadav

as i mentioned above comment follow the steps adding on top of it

Read Data from Event Hubs

Use Spark Streaming to read data from Event Hubs.

# Define Event Hubs configuration for reading event_hubs_conf = { 'eventhubs.connectionString': connection_string } # Read data from Event Hubs df = spark.readStream \ .format("eventhubs") \ .options(**event_hubs_conf) \ .load() # Process the data (example: print schema) df.printSchema() # Start the streaming query query = df.writeStream \ .outputMode("append") \ .format("console") \ .start() query.awaitTermination() ation()
Please sign in to rate this answer.
Deshpande, Chetan 0 Reputation points Microsoft Employee

2025-01-17T04:09:26.0133333+00:00

Have you tried this solution using MSI ? if you have can you please paste the results ... In my experience it does not work... are you sure that you can use writestream to write data to console in Synapse notebook ?

Haja Ghareeb Nawaz Mohamed Rabik 0 Reputation points Microsoft Employee

2025-01-31T02:19:14.4533333+00:00

@phemanth i dont think the below code works from Synapse notebook to get the token

credential = DefaultAzureCredential()

can you post any working code/logic to make the synapse notebook connect to event hubs using Managedidentity
Sign in to comment

Use comments to ask for clarification, additional information, or improvements to the question.

Share via

Connect to eventhub using Managed service identity (MSI) from Azure synapse analytics using pyspark spark streaming

1 answer

Your answer