[DOP-21445] remove hardcoding spark-version for recieving packages

Author: maxim-lixakov

poetry.lock

@@ -59,7 +59,9 @@ python-jose = { version = "^3.3.0", extras = ["cryptography"], optional = true }
 jinja2 = { version = "^3.1.4", optional = true }
 python-multipart = { version = ">=0.0.9,<0.0.20", optional = true }
 celery = { version = "^5.4.0", optional = true }
-onetl = { version = "^0.12.0", extras = ["spark"], optional = true }
+# spark-excel and spark-xml doesn't support Spark 3.5.2 and 3.5.3 yet
+pyspark = { version = "3.5.1", optional = true }
+onetl = { version = "^0.12.0", optional = true }
 pyyaml = {version = "*", optional = true}
 # due to not supporting MacOS 14.x https://www.psycopg.org/psycopg3/docs/news.html#psycopg-3-1-20
 psycopg = { version = ">=3.1.0,<3.2.4", extras = ["binary"], optional = true }
@@ -103,6 +105,7 @@ worker = [
     "sqlalchemy",
     "sqlalchemy-utils",
     "celery",
+    "pyspark",
     "onetl",
     "asgi-correlation-id",
     "jinja2",
@@ -135,7 +138,9 @@ pytest-asyncio = "^0.24.0"
 pytest-randomly = "^3.15.0"
 pytest-deadfixtures = "^2.2.1"
 pytest-mock = "^3.14.0"
-onetl = {extras = ["spark", "s3", "hdfs"], version = "^0.12.0"}
+# spark-excel and spark-xml doesn't support Spark 3.5.2 and 3.5.3 yet
+pyspark = { version = "3.5.1", optional = true }
+onetl = {extras = ["s3", "hdfs"], version = "^0.12.0"}
 faker = ">=28.4.1,<34.0.0"
 coverage = "^7.6.1"
 gevent = "^24.2.1"

pyproject.toml

@@ -32,16 +32,19 @@ def get_worker_spark_session(
         log.debug("Enabling Hive support")
         spark_builder = spark_builder.enableHiveSupport()
 
+    spark_builder.sparkContext.setLogLevel("DEBUG")
     return spark_builder.getOrCreate()
 
 
 def get_packages(db_type: str) -> list[str]:
+    import pyspark
     from onetl.connection import MSSQL, Clickhouse, MySQL, Oracle, Postgres, SparkS3
     from onetl.file.format import XML, Excel
 
+    spark_version = pyspark.__version__
     # see supported versions from https://mvnrepository.com/artifact/com.crealytics/spark-excel
-    file_formats_spark_packages: list[str] = XML.get_packages(spark_version="3.5.1") + Excel.get_packages(
-        spark_version="3.5.1",
+    file_formats_spark_packages: list[str] = XML.get_packages(spark_version=spark_version) + Excel.get_packages(
+        spark_version=spark_version,
     )
 
     if db_type == "postgres":

syncmaster/worker/spark.py

@@ -66,13 +66,11 @@ def spark(settings: Settings, request: FixtureRequest) -> SparkSession:
 
     spark = (
         SparkSession.builder.appName("celery_worker")
+        .enableHiveSupport()
         .config("spark.sql.pyspark.jvmStacktrace.enabled", "true")
         .config("spark.driver.host", "localhost")
     )
 
-    if "hive" in markers:
-        spark = spark.enableHiveSupport()
-
     if "postgres" in markers:
         maven_packages.extend(Postgres.get_packages())
 
@@ -116,9 +114,10 @@ def spark(settings: Settings, request: FixtureRequest) -> SparkSession:
         )
 
     if "hdfs" in markers or "s3" in markers:
+        spark_version = pyspark.__version__
         # see supported versions from https://mvnrepository.com/artifact/com.crealytics/spark-excel
-        file_formats_spark_packages: list[str] = XML.get_packages(spark_version="3.5.1") + Excel.get_packages(
-            spark_version="3.5.1",
+        file_formats_spark_packages: list[str] = XML.get_packages(spark_version=spark_version) + Excel.get_packages(
+            spark_version=spark_version,
         )
         maven_packages.extend(file_formats_spark_packages)
 

tests/test_integration/test_run_transfer/conftest.py

@@ -193,6 +193,11 @@ async def test_run_transfer_s3_to_postgres(
 @pytest.mark.parametrize(
     "target_file_format, file_format_flavor",
     [
+        pytest.param(
+            ("xml", {}),
+            "without_compression",
+            id="xml",
+        ),
         pytest.param(
             ("csv", {}),
             "with_header",
@@ -208,11 +213,6 @@ async def test_run_transfer_s3_to_postgres(
             "with_header",
             id="excel",
         ),
-        pytest.param(
-            ("xml", {}),
-            "without_compression",
-            id="xml",
-        ),
     ],
     indirect=["target_file_format", "file_format_flavor"],
 )