diff --git a/sdk/RELEASE.md b/sdk/RELEASE.md
index 0b2851eaaf6..231858833c2 100644
--- a/sdk/RELEASE.md
+++ b/sdk/RELEASE.md
@@ -4,6 +4,7 @@
 
 * Support passing parameters in v2 using google.protobuf.Value [\#6804](https://github.com/kubeflow/pipelines/pull/6804).
 * Implement experimental v2 `@component` component [\#6825](https://github.com/kubeflow/pipelines/pull/6825)
+* Add load_component_from_* for v2 [\#6822](https://github.com/kubeflow/pipelines/pull/6822)
 
 ## Breaking Changes
 
diff --git a/sdk/python/kfp/v2/components/experimental/test_data/simple_yaml.yaml b/sdk/python/kfp/v2/components/experimental/test_data/simple_yaml.yaml
new file mode 100644
index 00000000000..7035bcd07aa
--- /dev/null
+++ b/sdk/python/kfp/v2/components/experimental/test_data/simple_yaml.yaml
@@ -0,0 +1,16 @@
+name: component_1
+inputs:
+  input1: {type: String}
+outputs:
+  output1: {type: String}
+implementation:
+  container:
+    image: alpine
+    commands:
+    - sh
+    - -c
+    - 'set -ex
+
+    echo "$0" > "$1"'
+    - {inputValue: input1}
+    - {outputPath: output1}
\ No newline at end of file
diff --git a/sdk/python/kfp/v2/components/experimental/yaml_component.py b/sdk/python/kfp/v2/components/experimental/yaml_component.py
index 9cc855a9cd5..2e38e02cd4a 100644
--- a/sdk/python/kfp/v2/components/experimental/yaml_component.py
+++ b/sdk/python/kfp/v2/components/experimental/yaml_component.py
@@ -13,6 +13,12 @@
 # limitations under the License.
 """Functions for loading component from yaml."""
 
+__all__ = [
+    'load_component_from_text',
+    'load_component_from_url',
+    'load_component_from_file',
+]
+
 from kfp.v2.components.experimental import base_component
 from kfp.v2.components.experimental import structures
 
@@ -27,4 +33,34 @@ def execute(self, *args, **kwargs):
 def load_component_from_text(text: str) -> base_component.BaseComponent:
     """Loads component from text."""
     return YamlComponent(
-        component_spec=structures.ComponentSpec.load_from_component_yaml(text))
+        structures.ComponentSpec.load_from_component_yaml(text))
+
+def load_component_from_file(file_path: str) -> base_component.BaseComponent:
+    """Loads component from file.
+
+    Args:
+        file_path: A string containing path to the YAML file.
+    """
+    with open(file_path, 'rb') as component_stream:
+        return load_component_from_text(component_stream)
+
+def load_component_from_url(url: str, auth=None) -> base_component.BaseComponent:
+    """Loads component from url.
+
+    Args:
+        url: A string containing path to the url containing YAML file.
+        auth: The authentication credentials necessary for url access.
+    """
+
+    if url is None:
+        raise TypeError
+
+    if url.startswith('gs://'):
+        #Replacing the gs:// URI with https:// URI (works for public objects)
+        url = 'https://storage.googleapis.com/' + url[len('gs://'):]
+
+    import requests
+    resp = requests.get(url, auth=auth)
+    resp.raise_for_status()
+
+    return load_component_from_text(resp.content)
diff --git a/sdk/python/kfp/v2/components/experimental/yaml_component_test.py b/sdk/python/kfp/v2/components/experimental/yaml_component_test.py
new file mode 100644
index 00000000000..10668890f9e
--- /dev/null
+++ b/sdk/python/kfp/v2/components/experimental/yaml_component_test.py
@@ -0,0 +1,86 @@
+# Copyright 2021 The Kubeflow Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Tests for kfp.v2.components.experimental.yaml_component."""
+
+import requests
+import unittest
+import textwrap
+
+from pathlib import Path
+from unittest import mock
+
+from kfp.v2.components.experimental import yaml_component
+from kfp.v2.components.experimental import structures
+
+SAMPLE_YAML = textwrap.dedent("""\
+        name: component_1
+        inputs:
+          input1: {type: String}
+        outputs:
+          output1: {type: String}
+        implementation:
+          container:
+            image: alpine
+            commands:
+            - sh
+            - -c
+            - 'set -ex
+
+            echo "$0" > "$1"'
+            - {inputValue: input1}
+            - {outputPath: output1}
+        """)
+
+class YamlComponentTest(unittest.TestCase):
+
+    def test_load_component_from_text(self):
+        component = yaml_component.load_component_from_text(SAMPLE_YAML)
+        self.assertEqual(component.component_spec.name, 'component_1')
+        self.assertEqual(component.component_spec.outputs, {'output1': structures.OutputSpec(type='String')})
+        self.assertEqual(component._component_inputs, {'input1'})
+        self.assertEqual(component.name, 'component_1')
+        self.assertEqual(component.component_spec.implementation.container.image, 'alpine')
+
+    def test_load_component_from_file(self):
+        component_path = Path(
+            __file__).parent/'test_data'/'simple_yaml.yaml'
+        component = yaml_component.load_component_from_file(component_path)
+        self.assertEqual(component.component_spec.name, 'component_1')
+        self.assertEqual(component.component_spec.outputs, {'output1': structures.OutputSpec(type='String')})
+        self.assertEqual(component._component_inputs, {'input1'})
+        self.assertEqual(component.name, 'component_1')
+        self.assertEqual(component.component_spec.implementation.container.image, 'alpine')
+
+    def test_load_component_from_url(self):
+        component_url = 'https://raw.githubusercontent.com/some/repo/components/component_group/component.yaml'
+
+        def mock_response_factory(url, params=None, **kwargs):
+            if url == component_url:
+                response = requests.Response()
+                response.url = component_url
+                response.status_code = 200
+                response._content = SAMPLE_YAML
+                return response
+            raise RuntimeError('Unexpected URL "{}"'.format(url))
+
+        with mock.patch('requests.get', mock_response_factory):
+            component = yaml_component.load_component_from_url(component_url)
+            self.assertEqual(component.component_spec.name, 'component_1')
+            self.assertEqual(component.component_spec.outputs, {'output1': structures.OutputSpec(type='String')})
+            self.assertEqual(component._component_inputs, {'input1'})
+            self.assertEqual(component.name, 'component_1')
+            self.assertEqual(component.component_spec.implementation.container.image, 'alpine')
+
+if __name__ == '__main__':
+    unittest.main()