Merge pull request #74 from getindata/release-0.4.3

Release 0.4.3

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## [Unreleased]
 
+## [0.4.3] - 2021-09-27
+
+-   Kedro environment used by `kedro kubeflow` invocation is passed to the steps
+-   A flag to skip steps output artifacts registration in Kubeflow Metadata
+
 ## [0.4.2] - 2021-08-19
 
 -   Improved Vertex scheduling: removal of stale schedules
@@ -79,7 +84,9 @@
 -   Method to schedule runs for most recent version of given pipeline `kedro kubeflow schedule` 
 -   Shortcut to open UI for pipelines using `kedro kubeflow ui` 
 
-[Unreleased]: https://github.com/getindata/kedro-kubeflow/compare/0.4.2...HEAD
+[Unreleased]: https://github.com/getindata/kedro-kubeflow/compare/0.4.3...HEAD
+
+[0.4.3]: https://github.com/getindata/kedro-kubeflow/compare/0.4.2...0.4.3
 
 [0.4.2]: https://github.com/getindata/kedro-kubeflow/compare/0.4.1...0.4.2
 

kedro_kubeflow/__init__.py

@@ -1,3 +1,3 @@
 """kedro_kubeflow."""
 
-version = "0.4.2"
+version = "0.4.3"

kedro_kubeflow/config.py

@@ -40,6 +40,11 @@
   # See https://en.wikipedia.org/wiki/ISO_8601 in section 'Duration'
   #max_cache_staleness: P0D
 
+  # Set to false to disable kfp artifacts exposal
+  # This setting can be useful if you don't want to store
+  # intermediate results in the MLMD
+  #store_kedro_outputs_as_kfp_artifacts: True
+
   # Optional volume specification (only for non vertex-ai)
   volume:
 
@@ -195,6 +200,12 @@ def volume(self):
     def wait_for_completion(self):
         return bool(self._get_or_default("wait_for_completion", False))
 
+    @property
+    def store_kedro_outputs_as_kfp_artifacts(self):
+        return bool(
+            self._get_or_default("store_kedro_outputs_as_kfp_artifacts", True)
+        )
+
     @property
     def max_cache_staleness(self):
         return str(self._get_or_default("max_cache_staleness", None))

kedro_kubeflow/generator.py

@@ -138,6 +138,8 @@ def _build_kfp_ops(
                     command=["kedro"],
                     arguments=[
                         "kubeflow",
+                        "--env",
+                        self.context.env,
                         "mlflow-start",
                         dsl.RUN_ID_PLACEHOLDER,
                     ],
@@ -176,6 +178,8 @@ def _build_kfp_ops(
                     command=["kedro"],
                     arguments=[
                         "run",
+                        "--env",
+                        self.context.env,
                         "--params",
                         params,
                         "--pipeline",
@@ -191,6 +195,7 @@ def _build_kfp_ops(
                         for output in node.outputs
                         if output in self.catalog
                         and "filepath" in self.catalog[output]
+                        and self.run_config.store_kedro_outputs_as_kfp_artifacts
                     },
                 ),
                 image_pull_policy,

setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.4.2
+current_version = 0.4.3
 
 [bumpversion:file:setup.py]
 

setup.py

@@ -12,6 +12,7 @@
     "tabulate>=0.8.7",
     "semver~=2.10",
     "google-auth<2.0dev",
+    "protobuf<3.18.0",
 ]
 
 # Dev Requirements
@@ -36,7 +37,7 @@
 
 setup(
     name="kedro-kubeflow",
-    version="0.4.2",
+    version="0.4.3",
     description="Kedro plugin with Kubeflow support",
     long_description=README,
     long_description_content_type="text/markdown",

tests/test_generator.py

@@ -149,6 +149,8 @@ def test_should_add_mlflow_init_step_if_enabled(self):
         assert init_step.image == "unittest-image"
         assert init_step.args == [
             "kubeflow",
+            "--env",
+            "unittests",
             "mlflow-start",
             "{{workflow.uid}}",
         ]
@@ -201,6 +203,8 @@ def test_should_support_params_and_inject_them_to_the_nodes(self):
             args = dsl_pipeline.ops[node_name].container.args
             assert args == [
                 "run",
+                "--env",
+                "unittests",
                 "--params",
                 "param1:{{pipelineparam:op=;name=param1}},"
                 "param2:{{pipelineparam:op=;name=param2}}",
@@ -290,6 +294,29 @@ def test_artifact_registration(self):
         outputs2 = dsl_pipeline.ops["node2"].file_outputs
         assert len(outputs2) == 0  # output "C" is missing in the catalog
 
+    def test_should_skip_artifact_registration_if_requested(self):
+        # given
+        self.create_generator(
+            catalog={
+                "B": {
+                    "type": "pandas.CSVDataSet",
+                    "filepath": "data/02_intermediate/b.csv",
+                }
+            },
+            config={"store_kedro_outputs_as_kfp_artifacts": False},
+        )
+
+        # when
+        pipeline = self.generator_under_test.generate_pipeline(
+            "pipeline", "unittest-image", "Always"
+        )
+        with kfp.dsl.Pipeline(None) as dsl_pipeline:
+            pipeline()
+
+        # then
+        outputs1 = dsl_pipeline.ops["node1"].file_outputs
+        assert len(outputs1) == 0
+
     def test_should_skip_volume_removal_if_requested(self):
         # given
         self.create_generator(config={"volume": {"keep": True}})
@@ -312,6 +339,7 @@ def create_generator(self, config={}, params={}, catalog={}):
             "obj",
             (object,),
             {
+                "env": "unittests",
                 "params": params,
                 "config_loader": config_loader,
                 "pipelines": {