Repository: getindata/kedro-kubeflow
Pull request #71 (2021-08-31 13:23:11)

Ability to skip artifacts registration in Kubeflow Metadata
#### Description

If the pipeline has a lot of steps, their output can quickly fill Minio storage. This PR introduces an option to skip the automatic kedro outputs registration. If user wishes some artifact to be stored in the long term, it can be exposed as Mlflow artifact.

##### PR Checklist
- [x] Tests added
- [x] [Changelog](CHANGELOG.md) updated 



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 80755c6..874b2be 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 ## [Unreleased]
 
 -   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
 
diff --git a/kedro_kubeflow/cli.py b/kedro_kubeflow/cli.py
index 74211d4..213961d 100644
--- a/kedro_kubeflow/cli.py
+++ b/kedro_kubeflow/cli.py
@@ -22,7 +22,7 @@ def commands():
     name="kubeflow", context_settings=dict(help_option_names=["-h", "--help"])
 )
 @click.option(
-    "-e", "--env", "env", type=str, default="base", help="Environment to use."
+    "-e", "--env", "env", type=str, default="local", help="Environment to use."
 )
 @click.pass_obj
 @click.pass_context
diff --git a/kedro_kubeflow/config.py b/kedro_kubeflow/config.py
index 4c6e540..4ba8105 100644
--- a/kedro_kubeflow/config.py
+++ b/kedro_kubeflow/config.py
@@ -40,6 +40,11 @@ run_config:
   # 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 @@ class RunConfig(Config):
     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))
diff --git a/kedro_kubeflow/generator.py b/kedro_kubeflow/generator.py
index 06f1928..dc52862 100644
--- a/kedro_kubeflow/generator.py
+++ b/kedro_kubeflow/generator.py
@@ -195,6 +195,7 @@ class PipelineGenerator(object):
                         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,
diff --git a/setup.py b/setup.py
index dc28b6a..461226c 100644
--- a/setup.py
+++ b/setup.py
@@ -12,6 +12,7 @@ INSTALL_REQUIRES = [
     "tabulate>=0.8.7",
     "semver~=2.10",
     "google-auth<2.0dev",
+    "protobuf<3.18.0",
 ]
 
 # Dev Requirements


Test patch:
diff --git a/tests/test_generator.py b/tests/test_generator.py
index 57736be..92cf41a 100644
--- a/tests/test_generator.py
+++ b/tests/test_generator.py
@@ -294,6 +294,29 @@ class TestGenerator(unittest.TestCase):
         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}})