refactor: remove creating Spark session from Damavand

examples/sparkle/__main__.py

@@ -1,3 +1,4 @@
+import os
 from damavand.cloud.provider import AwsProvider
 from damavand.factories import SparkControllerFactory
 
@@ -16,14 +17,15 @@ def main() -> None:
 
     spark_controller = spark_factory.new(
         name="my-spark",
+        applications=[
+            Products(),
+            CustomerOrders(),
+        ],
     )
 
-    spark_controller.applications = [
-        Products(spark_controller.default_session()),
-        CustomerOrders(spark_controller.default_session()),
-    ]
+    app_name = os.getenv("APP_NAME", "default_app")  # Get app name on runtime
 
-    spark_controller.run_application("products")
+    spark_controller.run_application(app_name)
     spark_controller.provision()
 
 

examples/sparkle/applications/orders.py

@@ -1,30 +1,23 @@
 from sparkle.config import Config
 from sparkle.writer.iceberg_writer import IcebergWriter
 from sparkle.application import Sparkle
+from sparkle.reader.kafka_reader import KafkaReader
 
 from pyspark.sql import DataFrame
-from pyspark.sql import SparkSession
 
 
 class CustomerOrders(Sparkle):
-    def __init__(self, spark_session: SparkSession):
+    def __init__(self):
         super().__init__(
-            spark_session,
             config=Config(
                 app_name="orders",
                 app_id="orders-app",
                 version="0.0.1",
                 database_bucket="s3://test-bucket",
                 checkpoints_bucket="s3://test-checkpoints",
             ),
-            writers=[
-                IcebergWriter(
-                    database_name="default",
-                    database_path="s3://bucket-name/warehouse",
-                    table_name="products",
-                    spark_session=spark_session,
-                )
-            ],
+            readers={"orders": KafkaReader},
+            writers=[IcebergWriter],
         )
 
     def process(self) -> DataFrame:

examples/sparkle/applications/products.py

@@ -1,29 +1,26 @@
 from sparkle.application import Sparkle
 from sparkle.config import Config
 from sparkle.writer.iceberg_writer import IcebergWriter
+from sparkle.writer.kafka_writer import KafkaStreamPublisher
+from sparkle.reader.table_reader import TableReader
 
 from pyspark.sql import DataFrame
-from pyspark.sql import SparkSession
 
 
 class Products(Sparkle):
-    def __init__(self, spark_session: SparkSession):
+    def __init__(self):
         super().__init__(
-            spark_session,
             config=Config(
                 app_name="products",
                 app_id="products-app",
                 version="0.0.1",
                 database_bucket="s3://test-bucket",
                 checkpoints_bucket="s3://test-checkpoints",
             ),
+            readers={"products": TableReader},
             writers=[
-                IcebergWriter(
-                    database_name="default",
-                    database_path="s3://bucket-name/warehouse",
-                    table_name="products",
-                    spark_session=spark_session,
-                )
+                IcebergWriter,
+                KafkaStreamPublisher,
             ],
         )
 

pyproject.toml

@@ -13,7 +13,7 @@ dependencies = [
     "pulumi-aws>=6.47.0",
     "pulumi-azure-native>=2.51.0",
     "pulumi-random>=4.16.3",
-    "sparkle @ git+https://github.com/DataChefHQ/sparkle.git@v0.3.1",
+    "sparkle @ git+https://github.com/DataChefHQ/sparkle.git@v0.6.1",
 ]
 requires-python = ">=3.11.0"
 readme = "README.md"

src/damavand/base/controllers/spark.py

@@ -1,9 +1,5 @@
-import os
 import logging
-from pyspark.conf import SparkConf
-from pyspark.sql import SparkSession
 
-from damavand.environment import Environment
 from damavand.base.controllers import ApplicationController
 from damavand.base.controllers.base_controller import runtime
 
@@ -33,12 +29,6 @@ class SparkController(ApplicationController):
 
     Methods
     -------
-    default_local_session()
-        Return the default local Spark session.
-    default_cloud_session()
-        Return the default cloud Spark session.
-    default_session()
-        Return the currently active Spark session.
     application_with_id(app_id)
         Return the Spark application with the given ID.
     run_application(app_id)
@@ -48,86 +38,12 @@ class SparkController(ApplicationController):
     def __init__(
         self,
         name,
+        applications: list[Sparkle],
         tags: dict[str, str] = {},
         **kwargs,
     ) -> None:
         ApplicationController.__init__(self, name, tags, **kwargs)
-        self.applications: list[Sparkle] = []
-
-    @property
-    def _spark_extensions(self) -> list[str]:
-        return [
-            "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions",
-        ]
-
-    @property
-    def _spark_packages(self) -> list[str]:
-        return [
-            "org.apache.iceberg:iceberg-spark-runtime-3.2_2.12:1.3.1",
-            "org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0",
-            "org.apache.spark:spark-avro_2.12:3.3.0",
-        ]
-
-    def default_local_session(self) -> SparkSession:
-        """Return the default local Spark session.
-
-        Returns:
-            SparkSession: The Spark session.
-        """
-
-        ivy_settings_path = os.environ.get("IVY_SETTINGS_PATH", None)
-        LOCAL_CONFIG = {
-            "spark.sql.extensions": ",".join(self._spark_extensions),
-            "spark.jars.packages": ",".join(self._spark_packages),
-            "spark.sql.jsonGenerator.ignoreNullFields": False,
-            "spark.sql.session.timeZone": "UTC",
-            "spark.sql.catalog.spark_catalog": "org.apache.iceberg.spark.SparkSessionCatalog",
-            "spark.sql.catalog.spark_catalog.type": "hive",
-            "spark.sql.catalog.local": "org.apache.iceberg.spark.SparkCatalog",
-            "spark.sql.catalog.local.type": "hadoop",
-            "spark.sql.catalog.local.warehouse": "/tmp/warehouse",
-            "spark.sql.defaultCatalog": "local",
-        }
-
-        spark_conf = SparkConf()
-
-        for key, value in LOCAL_CONFIG.items():
-            spark_conf.set(key, str(value))
-
-        spark_session = (
-            # NOTE: Pyright does not work `@classproperty` decorator used in `SparkSession`. This however should be fixed in pyspark v4.
-            SparkSession.builder.master("local[*]")  # type: ignore
-            .appName("LocalDataProductApp")
-            .config(conf=spark_conf)
-        )
-
-        if ivy_settings_path:
-            spark_session.config("spark.jars.ivySettings", ivy_settings_path)
-
-        return spark_session.getOrCreate()
-
-    def default_cloud_session(self) -> SparkSession:
-        """Return the default Spark session provided by the cloud spark machine.
-
-        Returns:
-            SparkSession: The Spark session.
-        """
-
-        raise NotImplementedError
-
-    def default_session(self) -> SparkSession:
-        """Return the currently active Spark session. If the environment is local, it
-        returns the local session. Otherwise, it returns the cloud session.
-
-        Returns:
-            SparkSession: The Spark session.
-        """
-        env = Environment.from_system_env()
-        match env:
-            case Environment.LOCAL:
-                return self.default_local_session()
-            case _:
-                return self.default_cloud_session()
+        self.applications: list[Sparkle] = applications
 
     def application_with_id(self, app_id: str) -> Sparkle:
         """Return the Spark application with the given ID.

src/damavand/base/factory.py

@@ -1,9 +1,9 @@
 from dataclasses import field, dataclass
 from typing import Optional, Generic, TypeVar
 
+from sparkle.application import Sparkle
 from damavand.base.controllers import ApplicationController
 from damavand.cloud.provider import AwsProvider, AzurermProvider, CloudProvider
-
 from damavand.errors import UnsupportedProvider
 
 
@@ -36,13 +36,15 @@ class ApplicationControllerFactory(Generic[ControllerType]):
     def new(
         self,
         name: str,
+        applications: list[Sparkle],
         id: Optional[str] = None,
         **kwargs,
     ) -> ControllerType:
         match self.provider:
             case AwsProvider():
                 ctr = self._new_aws_controller(
                     name=name,
+                    applications=applications,
                     region=self.provider.explicit_region,
                     tags=self.tags,
                     **kwargs,
@@ -52,6 +54,7 @@ def new(
             case AzurermProvider():
                 ctr = self._new_azure_controller(
                     name=name,
+                    applications=applications,
                     tags=self.tags,
                     **kwargs,
                 )
@@ -63,6 +66,7 @@ def new(
     def _new_aws_controller(
         self,
         name: str,
+        applications: list[Sparkle],
         region: str,
         tags: dict[str, str] = {},
         **kwargs,
@@ -72,6 +76,7 @@ def _new_aws_controller(
     def _new_azure_controller(
         self,
         name: str,
+        applications: list[Sparkle],
         tags: dict[str, str] = {},
         **kwargs,
     ) -> ControllerType:

src/damavand/cloud/aws/controllers/spark.py

@@ -3,6 +3,7 @@
 
 import boto3
 from pulumi import Resource as PulumiResource
+from sparkle.application import Sparkle
 
 from damavand.base.controllers import SparkController, buildtime
 from damavand.cloud.aws.resources import GlueComponent, GlueComponentArgs
@@ -17,11 +18,12 @@ class AwsSparkController(SparkController):
     def __init__(
         self,
         name,
+        applications: list[Sparkle],
         region: str,
         tags: dict[str, str] = {},
         **kwargs,
     ) -> None:
-        super().__init__(name, tags, **kwargs)
+        super().__init__(name, applications, tags, **kwargs)
         self._glue_client = boto3.client("glue", region_name=region)
 
     @buildtime

src/damavand/cloud/azure/controllers/spark.py

@@ -18,12 +18,13 @@ class AzureSparkController(SparkController):
     def __init__(
         self,
         name,
+        applications: list[Sparkle],
         region: str,
         tags: dict[str, str] = {},
         **kwargs,
     ) -> None:
-        super().__init__(name, tags, **kwargs)
-        self.applications: list[Sparkle]
+        super().__init__(name, applications, tags, **kwargs)
+        self.applications = applications
 
     @buildtime
     def admin_username(self) -> str:

src/damavand/factories.py

@@ -1,4 +1,4 @@
-
+from sparkle.application import Sparkle
 from damavand.base.controllers.spark import SparkController
 from damavand.base.factory import ApplicationControllerFactory
 from damavand.cloud.aws.controllers import AwsSparkController
@@ -9,6 +9,7 @@ class SparkControllerFactory(ApplicationControllerFactory[SparkController]):
     def _new_aws_controller(
         self,
         name: str,
+        applications: list[Sparkle],
         region: str,
         tags: dict[str, str] = {},
         **kwargs,
@@ -21,7 +22,11 @@ def _new_aws_controller(
         )
 
     def _new_azure_controller(
-        self, name: str, tags: dict[str, str] = {}, **kwargs
+        self,
+        name: str,
+        applications: list[Sparkle],
+        tags: dict[str, str] = {},
+        **kwargs,
     ) -> SparkController:
         return AzureSparkController(
             name=name,

tests/base/test_spark.py

@@ -1,6 +1,4 @@
 import pytest
-from unittest.mock import patch
-from _pytest.monkeypatch import MonkeyPatch
 
 from damavand.base.controllers import SparkController
 
@@ -9,33 +7,5 @@
 def spark_controller():
     return SparkController(
         name="test",
+        applications=[],
     )
-
-
-def test_default_session(monkeypatch: MonkeyPatch, spark_controller: SparkController):
-    monkeypatch.setenv("ENVIRONMENT", "local")
-
-    with patch.object(spark_controller, "default_local_session") as mock_session:
-        spark_controller.default_session()
-        mock_session.assert_called_once()
-
-
-@pytest.mark.parametrize(
-    "environment",
-    [
-        "development",
-        "production",
-        "testing",
-        "acceptance",
-    ],
-)
-def test_default_session_with_environment(
-    monkeypatch: MonkeyPatch,
-    spark_controller: SparkController,
-    environment: str,
-):
-    monkeypatch.setenv("ENVIRONMENT", environment)
-
-    with patch.object(spark_controller, "default_cloud_session") as mock_session:
-        spark_controller.default_session()
-        mock_session.assert_called_once()

tests/clouds/aws/controllers/test_aws_spark.py

@@ -13,6 +13,7 @@
 def controller():
     ctr = AwsSparkController(
         "test-spark",
+        applications=[],
         region="us-east-1",
     )