pyspark - azure eventhub emulator - com.microsoft.azure.eventhubs.CommunicationException: Connection refused

[smpurkis]

Description

I'm using pyspark to write to eventhubs using this project. There is a new local eventhub emulator I'm using for testing locally https://github.com/Azure/azure-event-hubs-emulator-installer. But I get com.microsoft.azure.eventhubs.CommunicationException: Connection refused error when I try to use it with pyspark. It works if I use python azure eventhub packages.

Actual behavior

I get the error

Traceback (most recent call last):
  File "/Users/user/Projects/azure-event-hubs-emulator-installer/send_via_spark.py", line 76, in <module>
    serialized_rows.select("body").write.format("eventhubs").options(**ehConf).save()
  File "/Users/user/Projects/azure-event-hubs-emulator-installer/.venv/lib/python3.10/site-packages/pyspark/sql/readwriter.py", line 1461, in save
    self._jwrite.save()
  File "/Users/user/Projects/azure-event-hubs-emulator-installer/.venv/lib/python3.10/site-packages/py4j/java_gateway.py", line 1322, in __call__
    return_value = get_return_value(
  File "/Users/user/Projects/azure-event-hubs-emulator-installer/.venv/lib/python3.10/site-packages/pyspark/errors/exceptions/captured.py", line 179, in deco
    return f(*a, **kw)
  File "/Users/user/Projects/azure-event-hubs-emulator-installer/.venv/lib/python3.10/site-packages/py4j/protocol.py", line 326, in get_return_value
    raise Py4JJavaError(
py4j.protocol.Py4JJavaError: An error occurred while calling o86.save.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 6 in stage 4.0 failed 1 times, most recent failure: Lost task 6.0 in stage 4.0 (TID 26) (192.168.68.105 executor driver): com.microsoft.azure.eventhubs.CommunicationException: Connection refused

Expected behavior

I should be able to change the connection string and consumer group and it should work as if it is a remote/hosted/real eventhub.

Versions

Python: 3.10.13
Python virtualenv (pip freeze):

aiohappyeyeballs==2.4.0
aiohttp==3.10.5
aiosignal==1.3.1
async-timeout==4.0.3
attrs==24.2.0
azure-core==1.30.2
azure-eventhub==5.12.1
azure-eventhub-checkpointstoreblob-aio==1.1.4
certifi==2024.7.4
cffi==1.17.0
charset-normalizer==3.3.2
cryptography==43.0.0
frozenlist==1.4.1
idna==3.8
isodate==0.6.1
msrest==0.7.1
multidict==6.0.5
numpy==2.1.0
oauthlib==3.2.2
pandas==2.2.2
py4j==0.10.9.7
pycparser==2.22
pyspark==3.5.2
python-dateutil==2.9.0.post0
pytz==2024.1
requests==2.32.3
requests-oauthlib==2.0.0
six==1.16.0
typing_extensions==4.12.2
tzdata==2024.1
urllib3==2.2.2
wrapt==1.16.0
yarl==1.9.4

azure-event-hubs-spark: com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22

Reproduce

Working using azure eventhub python package

I use 3 scripts to test.

1. Get the local evenhub emulator running, following instructions via https://github.com/Azure/azure-event-hubs-emulator-installer repo.
2. Run listen.py in one terminal.

import asyncio

from azure.eventhub.aio import EventHubConsumerClient

EVENT_HUB_CONNECTION_STR = "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;"
EVENT_HUB_NAME = "eh1"
CONSUMER_GROUP = "cg1"


async def on_event(partition_context, event):
    # Print the event data.
    print(
        'Received the event: "{}" from the partition with ID: "{}"'.format(
            event.body_as_str(encoding="UTF-8"), partition_context.partition_id
        )
    )


async def main():

    # Create a consumer client for the event hub.
    client = EventHubConsumerClient.from_connection_string(
        EVENT_HUB_CONNECTION_STR,
        consumer_group=CONSUMER_GROUP,
        eventhub_name=EVENT_HUB_NAME,
    )
    async with client:
        # Call the receive method. Read from the beginning of the
        # partition (starting_position: "-1")
        await client.receive(on_event=on_event, starting_position="-1")


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    # Run the main method.
    loop.run_until_complete(main())

3. Run send.py in another terminal.

import asyncio

from azure.eventhub import EventData
from azure.eventhub.aio import EventHubProducerClient

EVENT_HUB_CONNECTION_STR = "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;"
EVENT_HUB_NAME = "eh1"


async def run():
    # Create a producer client to send messages to the event hub.
    # Specify a connection string to your event hubs namespace and
    # the event hub name.
    producer = EventHubProducerClient.from_connection_string(
        conn_str=EVENT_HUB_CONNECTION_STR, eventhub_name=EVENT_HUB_NAME
    )
    async with producer:
        # Create a batch.
        event_data_batch = await producer.create_batch()

        # Add events to the batch.
        event_data_batch.add(EventData("First event "))
        event_data_batch.add(EventData("Second event"))
        event_data_batch.add(EventData("Third event"))

        # Send the batch of events to the event hub.
        await producer.send_batch(event_data_batch)


asyncio.run(run())

4. In the listen.py terminal I receive events

> python listen.py
Received the event: "First event " from the partition with ID: "0"
Received the event: "Second event" from the partition with ID: "0"
Received the event: "Third event" from the partition with ID: "0"

Not working using pyspark and azure-event-hubs-spark

1. Get the local evenhub emulator running, following instructions via https://github.com/Azure/azure-event-hubs-emulator-installer repo.
2. Run the same listen.py script in one terminal.
3. Run send_via_spark.py in another terminal.

from pyspark.sql.functions import from_json
from pyspark.sql.types import StructField, StructType, StringType
from pyspark.sql import SparkSession
from pyspark.sql.functions import struct, to_json
from pyspark import SparkContext as sc


EVENT_HUB_NAME = "eh1"

EVENT_HUB_CONNECTION_STR = f"Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;EntityPath={EVENT_HUB_NAME}"


spark = (
    SparkSession.builder.appName("EventHubs")
    .config(
        "spark.jars.packages",
        "com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22,org.apache.hadoop:hadoop-azure:3.4.0",  # noqa
    )
    .config("spark.sql.shuffle.partitions", 5)
    .config("spark.hadoop.fs.azure", "org.apache.hadoop.fs.azure.NativeAzureFileSystem")
    .config("spark.hadoop.fs.azure.account.auth.type", "OAuth")
    .config(
        "spark.hadoop.fs.azure.account.oauth.provider.type",
        "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",  # noqa
    )
    .config("spark.hadoop.fs.azure.test.emulator", "true")
    .config(
        "fs.azure.storage.emulator.account.name",
        "devstoreaccount1.blob.windows.core.net",
    )
    .config(
        "spark.hadoop.fs.azure.storage.emulator.account.name",
        "devstoreaccount1.blob.windows.core.net",
    )
    .getOrCreate()
)


ehConf = {
    "eventhubs.connectionString": spark.sparkContext._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(
        EVENT_HUB_CONNECTION_STR
    )
}


# ehConf["eventhubs.consumerGroup"] = "$Default"
ehConf["eventhubs.consumerGroup"] = "cg1"

df = spark.createDataFrame(
    [
        ("1234", "1", "2021-01-01 00:00:00"),
        ("1234", "2", "2021-01-01 00:00:01"),
        ("1234", "3", "2021-01-01 00:00:02"),
    ],
    ["cardNumber", "transactionId", "transactionTime"],
)

Schema = StructType(
    [
        StructField("cardNumber", StringType(), True),
        StructField("transactionId", StringType(), True),
        StructField("transactionTime", StringType(), True),
    ]
)

serialized_rows = df.select(to_json(struct([df[x] for x in df.columns])).alias("body"))

# print(serialized_rows.select("body").toPandas())
serialized_rows.show(truncate=False)
print([r["body"] for r in serialized_rows.select("body").collect()])

print(ehConf)
serialized_rows.select("body").write.format("eventhubs").options(**ehConf).save()

4. In the listen.py terminal I receive nothing and in the send_via_spark.py terminal I get a rather lengthy error message
   error.log

[bingqiangxiang]

I also got the similar error when I used azure-eventhubs-spark to connect azure eventhub emulator,

[linomp]

Same for me in a Java-Spark project. I am using this package, which is the latest - but hasn't been updated since 2022:

<dependency>
    <groupId>com.microsoft.azure</groupId>
    <artifactId>azure-eventhubs-spark_2.12</artifactId>
    <version>2.3.22</version>
  </dependency>

My guess is that the client library needs to explicitly support the UseDevelopmentEmulator property, as shown in this snippet from the EventHubs .NET SDK source code.

And this is what they do after parsing the property, simply disabling TLS when using the emulator:

To test it, I tried to connect from a simple C# CLI app, with a connection string like this: "Endpoint=sb://localhost;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;EntityPath=eh1;UseDevelopmentEmulator=true" and it works well with the emulator, but if I omit the UseDevelopmentEmulator property the connection fails.

The azure-eventhubs-spark package is missing this explicit support! there has to be some way to force no-TLS for using the local emulator.