Complete Sparkle example

examples/sparkle/__main__.py

@@ -1,11 +1,17 @@
+from sparkle.application import Sparkle
 from damavand.cloud.provider import AwsProvider
 from damavand.factories import SparkControllerFactory
+from damavand.cloud.aws.resources.glue_component import (
+    GlueComponentArgs,
+    GlueJobDefinition,
+)
 
 from applications.orders import CustomerOrders
 from applications.products import Products
 
 
 def main() -> None:
+
     spark_factory = SparkControllerFactory(
         provider=AwsProvider(
             app_name="my-app",
@@ -14,12 +20,23 @@ def main() -> None:
         tags={"env": "dev"},
     )
 
+    applications: list[Sparkle] = [Products(), CustomerOrders()]
+    resource_args = GlueComponentArgs(
+        jobs=[
+            GlueJobDefinition(
+                name=app.config.app_name,
+                description=app.config.__doc__ or "",
+                script_location="__main__.py",
+                number_of_workers=2,
+            )
+            for app in applications
+        ],
+    )
+
     spark_controller = spark_factory.new(
         name="my-spark",
-        applications=[
-            Products(),
-            CustomerOrders(),
-        ],
+        applications=applications,
+        resource_args=resource_args,
     )
     # app_name = os.getenv("APP_NAME", "products-app")  # Get app name on runtime
 

examples/sparkle/build.sh

@@ -0,0 +1,5 @@
+#!/bin/bash
+
+pip install -r requirements.txt --target dependencies/
+python setup.py bdist_wheel --dist-dir artifacts
+aws s3 cp artifacts/damavand_packages-0.1-py3-none-any.whl s3://my-spark-code-bucket20241009085845037400000001

examples/sparkle/requirements.txt

@@ -1,3 +1,2 @@
--e ../../../damavand
-pyspark==3.3.2
-pulumi
+damavand @ git+https://github.com/DataChefHQ/damavand.git@1cf335462c873b5d3f938e1c58c7b7bc648ade74
+sparkle @ git+https://github.com/DataChefHQ/[email protected]

examples/sparkle/setup.py

@@ -0,0 +1,13 @@
+from setuptools import setup, find_packages
+
+setup(
+    name="damavand-packages",
+    version="0.1",
+    packages=find_packages(where="dependencies"),
+    package_dir={"": "dependencies"},
+    package_data={
+        "": ["*"],
+    },
+    include_package_data=True,
+    zip_safe=False,
+)

pyproject.toml

@@ -13,8 +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/[email protected]",
-    "damavand @ file:///${PROJECT_ROOT}/",
+    "sparkle @ git+https://github.com/DataChefHQ/[email protected]",
 ]
 requires-python = ">=3.11.0"
 readme = "README.md"

src/damavand/base/controllers/base_controller.py

@@ -1,5 +1,6 @@
 import logging
 from functools import cache
+from typing import Any
 from pulumi import Resource as PulumiResource
 import pulumi
 
@@ -41,6 +42,7 @@ def __init__(
         self,
         name: str,
         tags: dict[str, str] = {},
+        resource_args: Any = None,
         **kwargs,
     ) -> None:
         self.name = name

src/damavand/base/controllers/spark.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Any
 
 from damavand.base.controllers import ApplicationController
 from damavand.base.controllers.base_controller import runtime
@@ -24,6 +25,8 @@ class SparkController(ApplicationController):
         the list of Spark applications.
     tags : dict[str, str]
         the tags of the controller.
+    resource_args : Any
+        Any extra arguments for the underlying resource.
     kwargs : dict
         the extra arguments.
 
@@ -40,6 +43,7 @@ def __init__(
         name,
         applications: list[Sparkle],
         tags: dict[str, str] = {},
+        resource_args: Any = None,
         **kwargs,
     ) -> None:
         ApplicationController.__init__(self, name, tags, **kwargs)

src/damavand/base/factory.py

@@ -1,5 +1,5 @@
 from dataclasses import field, dataclass
-from typing import Optional, Generic, TypeVar
+from typing import Optional, Generic, TypeVar, Any
 
 from sparkle.application import Sparkle
 from damavand.base.controllers import ApplicationController
@@ -22,6 +22,8 @@ class ApplicationControllerFactory(Generic[ControllerType]):
         The cloud provider object.
     tags : dict[str, str]
         A set of default tags to be applied to all resources.
+    resource_args: any
+        Any extra arguments for the underlying resource.
 
     Methods
     -------
@@ -38,6 +40,7 @@ def new(
         name: str,
         applications: list[Sparkle],
         id: Optional[str] = None,
+        resource_args: Any = None,
         **kwargs,
     ) -> ControllerType:
         match self.provider:
@@ -47,6 +50,7 @@ def new(
                     applications=applications,
                     region=self.provider.explicit_region,
                     tags=self.tags,
+                    resource_args=resource_args,
                     **kwargs,
                 )
 
@@ -56,6 +60,7 @@ def new(
                     name=name,
                     applications=applications,
                     tags=self.tags,
+                    resource_args=resource_args,
                     **kwargs,
                 )
 
@@ -69,6 +74,7 @@ def _new_aws_controller(
         applications: list[Sparkle],
         region: str,
         tags: dict[str, str] = {},
+        resource_args: Any = None,
         **kwargs,
     ) -> ControllerType:
         raise NotImplementedError()
@@ -78,6 +84,7 @@ def _new_azure_controller(
         name: str,
         applications: list[Sparkle],
         tags: dict[str, str] = {},
+        resource_args: Any = None,
         **kwargs,
     ) -> ControllerType:
         raise NotImplementedError()

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

@@ -1,14 +1,14 @@
 import logging
 from functools import cache
+from typing import Any
 
 import boto3
 from pulumi import Resource as PulumiResource
 from sparkle.application import Sparkle
 
 from damavand.base.controllers import buildtime
 from damavand.base.controllers.spark import SparkController
-from damavand.cloud.aws.resources import GlueComponent, GlueComponentArgs
-from damavand.cloud.aws.resources.glue_component import GlueJobDefinition
+from damavand.cloud.aws.resources import GlueComponent
 from damavand.errors import BuildtimeException
 
 
@@ -22,26 +22,17 @@ def __init__(
         applications: list[Sparkle],
         region: str,
         tags: dict[str, str] = {},
+        resource_args: Any = None,
         **kwargs,
     ) -> None:
         super().__init__(name, applications, tags, **kwargs)
         self._glue_client = boto3.client("glue", region_name=region)
+        self.resource_args = resource_args
 
     @buildtime
     @cache
     def resource(self) -> PulumiResource:
         if not self.applications:
             raise BuildtimeException("No applications found to create Glue jobs.")
 
-        return GlueComponent(
-            name=self.name,
-            args=GlueComponentArgs(
-                jobs=[
-                    GlueJobDefinition(
-                        name=app.config.app_name,
-                        description=app.config.__doc__ or "",
-                    )
-                    for app in self.applications
-                ],
-            ),
-        )
+        return GlueComponent(name=self.name, args=self.resource_args)

src/damavand/cloud/aws/resources/glue_component.py

@@ -7,7 +7,7 @@
 
 import pulumi_aws as aws
 from pulumi import ComponentResource as PulumiComponentResource
-from pulumi import ResourceOptions
+from pulumi import ResourceOptions, Output, asset
 
 
 class GlueWorkerType(Enum):
@@ -66,8 +66,9 @@ class GlueJobDefinition:
 
     Attributes:
         name (str): Job name.
+        script_location (str): Local path to the entrypoint script.
+
         description (str): Job description.
-        script_location (str): S3 path to the entrypoint script.
         extra_libraries (list[str]): Paths to extra dependencies.
         execution_class (GlueExecutionClass): Execution class. Default is STANDARD.
         max_concurrent_runs (int): Max concurrent runs.
@@ -89,8 +90,8 @@ class GlueJobDefinition:
     """
 
     name: str
+    script_location: str
     description: str = ""
-    script_location: str = ""
     extra_libraries: list[str] = field(default_factory=list)
     execution_class: GlueExecutionClass = GlueExecutionClass.STANDARD
     max_concurrent_runs: int = 1
@@ -222,18 +223,23 @@ def _get_job_name(self, job: GlueJobDefinition) -> str:
         return f"{self._name}-{job.name}-job"
 
     def _get_source_path(self, job: GlueJobDefinition) -> str:
-        """Gets the source path for the job script.
+        """Uploads the job script to S3 and returns the S3 path.
 
         Args:
             job (GlueJobDefinition): The job definition.
 
         Returns:
             str: The S3 path to the job script.
         """
-        return (
-            f"s3://{self.code_repository_bucket.bucket}/{job.script_location}"
-            if job.script_location
-            else f"s3://{self.code_repository_bucket.bucket}/{job.name}.py"
+        # Upload Glue script to S3 bucket
+        glue_script = aws.s3.BucketObject(
+            f"{job.name}-entry-object",
+            bucket=self.code_repository_bucket.id,
+            source=asset.FileAsset(job.script_location),
+            key=f"{job.name}_main.py",
+        )
+        return Output.concat(
+            "s3://", self.code_repository_bucket.bucket, "/", glue_script.key
         )
 
     def _get_default_arguments(self, job: GlueJobDefinition) -> dict[str, str]:

src/damavand/factories.py

@@ -1,4 +1,5 @@
 from sparkle.application import Sparkle
+from typing import Any
 from damavand.base.controllers.spark import SparkController
 from damavand.base.factory import ApplicationControllerFactory
 from damavand.cloud.aws.controllers.spark import AwsSparkController
@@ -12,13 +13,15 @@ def _new_aws_controller(
         applications: list[Sparkle],
         region: str,
         tags: dict[str, str] = {},
+        resource_args: Any = None,
         **kwargs,
     ) -> SparkController:
         return AwsSparkController(
             name=name,
             applications=applications,
             region=region,
             tags=tags,
+            resource_args=resource_args,
             **kwargs,
         )
 
@@ -27,6 +30,7 @@ def _new_azure_controller(
         name: str,
         applications: list[Sparkle],
         tags: dict[str, str] = {},
+        resource_args: Any = None,
         **kwargs,
     ) -> SparkController:
         return AzureSparkController(