Merge branch 'feature/spark-declarative-pipelines-component' of https://github.com/michalcabir-ui/dagster into feature/spark-declarative-pipelines-component

Author: michalcabir-ui

python_modules/libraries/dagster-spark/dagster_spark/components/spark_declarative_pipeline/component.py

@@ -42,7 +42,12 @@ def _resolve_spark_pipelines_resource(_context: Any, value: Any) -> SparkPipelin
         return value
     if value is None:
         return SparkPipelinesResource()
-    return SparkPipelinesResource(**value) if isinstance(value, dict) else SparkPipelinesResource()
+    if isinstance(value, dict):
+        return SparkPipelinesResource(**value)
+    raise ValueError(
+        f"Cannot resolve spark_pipelines field: expected a SparkPipelinesResource, dict, or None, "
+        f"got {type(value).__name__!r}: {value!r}"
+    )
 
 
 @scaffold_with(SparkDeclarativePipelineScaffolder)

python_modules/libraries/dagster-spark/dagster_spark/components/spark_declarative_pipeline/resource.py

@@ -84,7 +84,8 @@ def run_and_observe(
 
         Uses Popen to stream stdout/stderr line-by-line and logs each line via context.log.info.
         Passes --full-refresh or --refresh based on execution_mode, then optional comma-separated
-        dataset list from asset_keys.         returncode == 0; otherwise raises SparkPipelinesExecutionError with the captured log.
+        dataset list from asset_keys. Yields MaterializeResult per asset on success when
+        returncode == 0; otherwise raises SparkPipelinesExecutionError with the captured log.
 
         Args:
             context: Asset execution context (used for context.log.info).
@@ -133,14 +134,16 @@ def run_and_observe(
             bufsize=1,
         )
         log_lines: deque[str] = deque(maxlen=1000)
-        if process.stdout:
-            for raw_line in process.stdout:
-                line = raw_line.rstrip("\n\r")
-                if line:
-                    log_lines.append(line)
-                    if context is not None and hasattr(context, "log"):
-                        context.log.info(line)
-        process.wait()
+        try:
+            if process.stdout:
+                for raw_line in process.stdout:
+                    line = raw_line.rstrip("\n\r")
+                    if line:
+                        log_lines.append(line)
+                        if context is not None and hasattr(context, "log"):
+                            context.log.info(line)
+        finally:
+            process.wait()
         returncode = process.returncode
 
         if returncode != 0: