Merge pull request #259 from m-aciek/spark-4-support

Add Spark 4.0 support via deequ:2.0.14-spark-4.0

Author: sudsali

.github/workflows/base.yml

@@ -25,6 +25,10 @@ jobs:
           - PYSPARK_VERSION: "3.5"
             PYTHON_VERSION: "3.9"
             JAVA_VERSION: "17"
+          - PYSPARK_VERSION: "4.0"
+            PYTHON_VERSION: "3.9"
+            JAVA_VERSION: "17"
+            PANDAS_VERSION_SPEC: ">=2.0.0"
 
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
@@ -41,9 +45,11 @@ jobs:
       - name: Running tests with pyspark==${{matrix.PYSPARK_VERSION}}
         env:
           SPARK_VERSION: ${{matrix.PYSPARK_VERSION}}
+          PANDAS_VERSION_SPEC: ${{matrix.PANDAS_VERSION_SPEC}}
         run: |
           pip install --upgrade pip
           pip install poetry==1.7.1
           poetry install
           poetry run pip install pyspark==$SPARK_VERSION
+          if [ -n "$PANDAS_VERSION_SPEC" ]; then poetry run pip install "pandas$PANDAS_VERSION_SPEC"; fi
           poetry run python -m pytest -s tests --ignore=tests/test_bot.py

pydeequ/analyzers.py

@@ -9,7 +9,7 @@
 from pydeequ.pandas_utils import ensure_pyspark_df
 from pydeequ.repository import MetricsRepository, ResultKey
 from enum import Enum
-from pydeequ.scala_utils import to_scala_seq
+from pydeequ.scala_utils import empty_scala_seq, to_scala_seq
 from pydeequ.configs import SPARK_VERSION
 
 class _AnalyzerObject:
@@ -311,7 +311,7 @@ def _analyzer_jvm(self):
             self.instance,
             self.predicate,
             self._jvm.scala.Option.apply(self.where),
-            self._jvm.scala.collection.Seq.empty(),
+            empty_scala_seq(self._jvm),
             self._jvm.scala.Option.apply(None)
         )
 

pydeequ/checks.py

@@ -5,7 +5,7 @@
 from pyspark.sql import SparkSession
 
 from pydeequ.check_functions import is_one
-from pydeequ.scala_utils import ScalaFunction1, to_scala_seq
+from pydeequ.scala_utils import ScalaFunction1, empty_scala_seq, to_scala_seq
 from pydeequ.configs import SPARK_VERSION
 
 # TODO implement custom assertions
@@ -563,7 +563,7 @@ def satisfies(self, columnCondition, constraintName, assertion=None, hint=None):
             constraintName,
             assertion_func,
             hint,
-            self._jvm.scala.collection.Seq.empty(),
+            empty_scala_seq(self._jvm),
             self._jvm.scala.Option.apply(None)
         )
         return self

pydeequ/configs.py

@@ -5,6 +5,7 @@
 
 
 SPARK_TO_DEEQU_COORD_MAPPING = {
+    "4.0": "com.amazon.deequ:deequ:2.0.14-spark-4.0",
     "3.5": "com.amazon.deequ:deequ:2.0.8-spark-3.5",
     "3.3": "com.amazon.deequ:deequ:2.0.8-spark-3.3",
     "3.2": "com.amazon.deequ:deequ:2.0.8-spark-3.2",

pydeequ/profiles.py

@@ -254,7 +254,7 @@ def _columnProfilesFromColumnRunBuilderRun(self, run):
         :return: a setter for columnProfilerRunner result
         """
         self._run_result = run
-        profile_map = self._jvm.scala.collection.JavaConversions.mapAsJavaMap(run.profiles())  # TODO from ScalaUtils
+        profile_map = scala_map_to_java_map(self._jvm, run.profiles())
         self._profiles = {column: self._columnProfileBuilder(column, profile_map[column]) for column in profile_map}
         return self
 

pydeequ/scala_utils.py

@@ -68,6 +68,30 @@ def get_or_else_none(scala_option):
     return scala_option.get()
 
 
+# Cache per JVM instance so version detection only happens once per session.
+_jvm_converters_cache: dict = {}
+
+
+def _get_converters(jvm):
+    """
+    Return (style, converters) for the running Scala version.
+    style='jdk'    → scala.jdk.javaapi.CollectionConverters (Scala 2.13, Spark 4+)
+    style='legacy' → scala.collection.JavaConverters        (Scala 2.12, Spark 3.x)
+    """
+    key = id(jvm)
+    if key not in _jvm_converters_cache:
+        try:
+            converters = jvm.scala.jdk.javaapi.CollectionConverters
+            # On Scala 2.12, the path resolves to a JavaPackage placeholder (no class
+            # exists), so attribute access succeeds but any method call raises TypeError.
+            # Probe with an actual call to confirm the class is genuinely usable.
+            converters.asScala(jvm.java.util.ArrayList())
+            _jvm_converters_cache[key] = ("jdk", converters)
+        except Exception:
+            _jvm_converters_cache[key] = ("legacy", jvm.scala.collection.JavaConverters)
+    return _jvm_converters_cache[key]
+
+
 def to_scala_seq(jvm, iterable):
     """
     Helper method to take an iterable and turn it into a Scala sequence
@@ -77,7 +101,23 @@ def to_scala_seq(jvm, iterable):
     Returns:
         Scala sequence
     """
-    return jvm.scala.collection.JavaConversions.iterableAsScalaIterable(iterable).toSeq()
+    style, converters = _get_converters(jvm)
+    if style == "jdk":
+        return converters.asScala(iterable).toSeq()
+    return converters.iterableAsScalaIterableConverter(iterable).asScala().toSeq()
+
+
+def empty_scala_seq(jvm):
+    """
+    Returns an empty Scala immutable List (Nil), usable as Seq[_].
+    Converts an empty ArrayList via .asScala().toList() to produce an immutable.List
+    rather than a Stream, which is required for Py4J constructor/method lookup to
+    succeed across both Scala 2.12 (Spark 3.x) and Scala 2.13 (Spark 4+).
+    """
+    style, converters = _get_converters(jvm)
+    if style == "jdk":
+        return converters.asScala(jvm.java.util.ArrayList()).toList()
+    return converters.iterableAsScalaIterableConverter(jvm.java.util.ArrayList()).asScala().toList()
 
 
 def to_scala_map(spark_session, d):
@@ -89,15 +129,29 @@ def to_scala_map(spark_session, d):
     Returns:
         Scala map
     """
-    return spark_session._jvm.PythonUtils.toScalaMap(d)
+    jvm = spark_session._jvm
+    try:
+        # PythonUtils.toScalaMap is a PySpark internal that may be removed in future versions.
+        return jvm.PythonUtils.toScalaMap(d)
+    except Exception:
+        style, converters = _get_converters(jvm)
+        if style == "jdk":
+            return converters.asScala(d).toMap()
+        return converters.mapAsScalaMapConverter(d).asScala().toMap()
 
 
 def scala_map_to_dict(jvm, scala_map):
-    return dict(jvm.scala.collection.JavaConversions.mapAsJavaMap(scala_map))
+    style, converters = _get_converters(jvm)
+    if style == "jdk":
+        return dict(converters.asJava(scala_map))
+    return dict(converters.mapAsJavaMapConverter(scala_map).asJava())
 
 
 def scala_map_to_java_map(jvm, scala_map):
-    return jvm.scala.collection.JavaConversions.mapAsJavaMap(scala_map)
+    style, converters = _get_converters(jvm)
+    if style == "jdk":
+        return converters.asJava(scala_map)
+    return converters.mapAsJavaMapConverter(scala_map).asJava()
 
 
 def java_list_to_python_list(java_list: str, datatype):

pyproject.toml

@@ -31,7 +31,7 @@ classifiers = [
 python = ">=3.8,<4"
 numpy = ">=1.14.1"
 pandas = ">=0.23.0"
-pyspark = { version = ">=2.4.7,<4.0.0", optional = true }
+pyspark = { version = ">=2.4.7,<5.0.0", optional = true }
 
 [tool.poetry.dev-dependencies]
 pytest = "^6.2.4"