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

.github/workflows/base.yml

@@ -12,7 +12,17 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        PYSPARK_VERSION: ["3.1.3", "3.2", "3.3", "3.5"]
+        include:
+          - PYSPARK_VERSION: "3.1.3"
+            JAVA_VERSION: "11"
+          - PYSPARK_VERSION: "3.2"
+            JAVA_VERSION: "11"
+          - PYSPARK_VERSION: "3.3"
+            JAVA_VERSION: "11"
+          - PYSPARK_VERSION: "3.5"
+            JAVA_VERSION: "11"
+          - PYSPARK_VERSION: "4.0.0"
+            JAVA_VERSION: "17"
 
     steps:
       - uses: actions/checkout@v3
@@ -23,10 +33,9 @@ jobs:
           python-version: 3.8
 
       - uses: actions/setup-java@v1
-        name: Setup Java 11
-        if: startsWith(matrix.PYSPARK_VERSION, '3')
+        name: Setup Java ${{matrix.JAVA_VERSION}}
         with:
-          java-version: "11"
+          java-version: ${{matrix.JAVA_VERSION}}
 
       - name: Running tests with pyspark==${{matrix.PYSPARK_VERSION}}
         env:

pydeequ/analyzers.py

@@ -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(),
+            to_scala_seq(self._jvm, self._jvm.java.util.ArrayList()),
             self._jvm.scala.Option.apply(None)
         )
 

pydeequ/checks.py

@@ -563,7 +563,7 @@ def satisfies(self, columnCondition, constraintName, assertion=None, hint=None):
             constraintName,
             assertion_func,
             hint,
-            self._jvm.scala.collection.Seq.empty(),
+            to_scala_seq(self._jvm, self._jvm.java.util.ArrayList()),
             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 = self._jvm.scala.collection.JavaConverters.mapAsJavaMapConverter(run.profiles()).asJava()  # TODO from ScalaUtils
         self._profiles = {column: self._columnProfileBuilder(column, profile_map[column]) for column in profile_map}
         return self
 

pydeequ/scala_utils.py

@@ -77,7 +77,7 @@ def to_scala_seq(jvm, iterable):
     Returns:
         Scala sequence
     """
-    return jvm.scala.collection.JavaConversions.iterableAsScalaIterable(iterable).toSeq()
+    return jvm.scala.collection.JavaConverters.iterableAsScalaIterableConverter(iterable).asScala().toSeq()
 
 
 def to_scala_map(spark_session, d):
@@ -93,11 +93,11 @@ def to_scala_map(spark_session, d):
 
 
 def scala_map_to_dict(jvm, scala_map):
-    return dict(jvm.scala.collection.JavaConversions.mapAsJavaMap(scala_map))
+    return dict(jvm.scala.collection.JavaConverters.mapAsJavaMapConverter(scala_map).asJava())
 
 
 def scala_map_to_java_map(jvm, scala_map):
-    return jvm.scala.collection.JavaConversions.mapAsJavaMap(scala_map)
+    return jvm.scala.collection.JavaConverters.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,<3.4.0", optional = true }
+pyspark = { version = ">=2.4.7,<5.0.0", optional = true }
 
 [tool.poetry.dev-dependencies]
 pytest = "^6.2.4"