Running multiple pydeequ checks at the same time in emr cluster OSError: [Errno 98] Address already in use :(127.0.0.1:25334)

[Ashokgoa]

Describe the bug
How did you handle this issue if you run many spark jobs which is running pydeequ checks at the same time?

in my case only one job is running rest of the other jobs are failing with this issue when the other jobs are running in the emr cluster.

Traceback (most recent call last):
File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 2207, in start
OSError: [Errno 98] Address already in use

During handling of the above exception, another exception occurred: Any solutions or suggestions

Traceback (most recent call last):
File "/opt/ammar/pydeequ_poc_pyspark.py", line 26, in
_check = _check_func(*_args)
File "/usr/local/lib/python3.7/site-packages/pydeequ/checks.py", line 134, in hasSize
assertion_func = ScalaFunction1(self._spark_session.sparkContext._gateway, assertion)
File "/usr/local/lib/python3.7/site-packages/pydeequ/scala_utils.py", line 32, in init
super().init(gateway)
File "/usr/local/lib/python3.7/site-packages/pydeequ/scala_utils.py", line 16, in init
self.gateway.start_callback_server()
File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 1894, in start_callback_server
File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 2216, in start
py4j.protocol.Py4JNetworkError: An error occurred while trying to start the callback server (127.0.0.1:25334)
21/11/19 13:44:06 INFO SparkContext: Invoking stop() from shutdown hookcise description of what the bug is.

To Reproduce
Steps to reproduce the behavior:
Run multiple pydeequ checks at the same time in an emr cluster

[chenliu0831]

Please provide a minimal code snippet if possible. Thank you

[Ashokgoa]

Hi Chenliu,

currenlty i have multiple airlfow jobs which are triggered at the same time for different spark jobs to emr cluster

in each of these airflow dags we added a step for data quality checks and in some cases when these data quality checks are running at the same time in cluster. the error occurs
this piece of code:
check = Check(spark, CheckLevel.Error, "Integrity checks")

checkResult = VerificationSuite(spark)
.onData(df)
.addCheck(
check
.isComplete("SubID")
.hasCompleteness("ErrorCode_",lambda x: x < 0.2)
.hasDataType("JobID",ConstrainableDataTypes.Integral)
.hasMaxLength("JobID",lambda x: x == 8)
.hasDataType("ListID",ConstrainableDataTypes.Integral)
.containsEmail("Emailaddr")
.hasMaxLength("ListID",lambda x: x == 5))
.useRepository(repository)
.saveOrAppendResult(resultKey)
.run()

most of the jobs fail with error:
Traceback (most recent call last):
File "/usr/lib/spark/python/lib/py4j-0.10.9-src.zip/py4j/java_gateway.py", line 2207, in start
OSError: [Errno 98] Address already in use

i can understand that the pydeeque always listens to 25334 port which causing other jobs to fail as they run same time.

please refer to this #19

[chenliu0831]

Thanks for providing more details and the related issue, will research this a bit.

[Ashokgoa]

can we implement this to make the dq to use next available not 25334
https://www.py4j.org/advanced_topics.html#using-py4j-without-pre-determined-ports-dynamic-port-number

[github-actions]

This is a known limitation. The issue is in PythonCallback.__init__ in scala_utils.py — when gateway.start_callback_server() is called, it binds to a fixed port (default 25334). When multiple Spark jobs run concurrently on the same node, the second job fails because that port is already in use.

As you correctly identified, the fix would be to use dynamic port allocation. Here's a workaround you can apply by patching the callback server to use port 0 (OS-assigned):

from pyspark.sql import SparkSession
from py4j.java_gateway import CallbackServerParameters

spark = SparkSession.builder.getOrCreate()
gateway = spark.sparkContext._gateway
gateway.start_callback_server(
    callback_server_parameters=CallbackServerParameters(port=0)
)

Call this before any PyDeequ check methods (like hasSize, hasCompleteness, etc.) so that when PythonCallback.__init__ runs, it sees the callback server is already started and skips the conflicting start_callback_server() call.

The proper fix in the library would be to update PythonCallback.__init__ to pass port=0 for dynamic port assignment:

from py4j.java_gateway import CallbackServerParameters

class PythonCallback:
    def __init__(self, gateway):
        self.gateway = gateway
        callback_server = self.gateway.get_callback_server()
        if callback_server is None:
            self.gateway.start_callback_server(
                callback_server_parameters=CallbackServerParameters(port=0)
            )
        elif callback_server.is_shutdown:
            callback_server.close()
            self.gateway.restart_callback_server()

We'll look into incorporating this fix. In the meantime, the pre-start workaround above should unblock you.

---

Generated by AI (model: us.anthropic.claude-opus-4-6-v1, prompt: cf3780e5) — may not be fully accurate. Reply if this doesn't help.