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

@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.