diff --git a/airflow/providers/cncf/kubernetes/operators/resource.py b/airflow/providers/cncf/kubernetes/operators/resource.py
new file mode 100644
index 0000000000000..9515e5b7e5e56
--- /dev/null
+++ b/airflow/providers/cncf/kubernetes/operators/resource.py
@@ -0,0 +1,103 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+"""Manage a Kubernetes Resource"""
+
+from __future__ import annotations
+
+from functools import cached_property
+
+import yaml
+from kubernetes.client import ApiClient
+from kubernetes.utils import create_from_yaml
+
+from airflow.models import BaseOperator
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.utils.delete_from import delete_from_yaml
+
+__all__ = ["KubernetesCreateResourceOperator", "KubernetesDeleteResourceOperator"]
+
+
+class KubernetesResourceBaseOperator(BaseOperator):
+    """
+    Abstract base class for all Kubernetes Resource operators.
+
+    :param yaml_conf: string. Contains the kubernetes resources to Create or Delete
+    :param namespace: string. Contains the namespace to create all resources inside.
+        The namespace must preexist otherwise the resource creation will fail.
+        If the API object in the yaml file already contains a namespace definition then
+        this parameter has no effect.
+    :param kubernetes_conn_id: The :ref:`kubernetes connection id <howto/connection:kubernetes>`
+        for the Kubernetes cluster.
+    :param in_cluster: run kubernetes client with in_cluster configuration.
+    :param cluster_context: context that points to kubernetes cluster.
+        Ignored when in_cluster is True. If None, current-context is used.
+    :param config_file: The path to the Kubernetes config file. (templated)
+        If not specified, default value is ``~/.kube/config``
+    """
+
+    template_fields = ("yaml_conf",)
+    template_fields_renderers = {"yaml_conf": "yaml"}
+
+    def __init__(
+        self,
+        *,
+        yaml_conf: str,
+        namespace: str | None = None,
+        kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+        **kwargs,
+    ) -> None:
+        super().__init__(**kwargs)
+        self._namespace = namespace
+        self.kubernetes_conn_id = kubernetes_conn_id
+        self.yaml_conf = yaml_conf
+
+    @cached_property
+    def client(self) -> ApiClient:
+        return self.hook.api_client
+
+    @cached_property
+    def hook(self) -> KubernetesHook:
+        hook = KubernetesHook(conn_id=self.kubernetes_conn_id)
+        return hook
+
+    def get_namespace(self) -> str:
+        if self._namespace:
+            return self._namespace
+        else:
+            return self.hook.get_namespace() or "default"
+
+
+class KubernetesCreateResourceOperator(KubernetesResourceBaseOperator):
+    """Create a resource in a kubernetes."""
+
+    def execute(self, context) -> None:
+        create_from_yaml(
+            k8s_client=self.client,
+            yaml_objects=yaml.safe_load_all(self.yaml_conf),
+            namespace=self.get_namespace(),
+        )
+
+
+class KubernetesDeleteResourceOperator(KubernetesResourceBaseOperator):
+    """Delete a resource in a kubernetes."""
+
+    def execute(self, context) -> None:
+        delete_from_yaml(
+            k8s_client=self.client,
+            yaml_objects=yaml.safe_load_all(self.yaml_conf),
+            namespace=self.get_namespace(),
+        )
diff --git a/airflow/providers/cncf/kubernetes/provider.yaml b/airflow/providers/cncf/kubernetes/provider.yaml
index 590df0c1c8bd1..ee8d9fecffc67 100644
--- a/airflow/providers/cncf/kubernetes/provider.yaml
+++ b/airflow/providers/cncf/kubernetes/provider.yaml
@@ -96,6 +96,7 @@ operators:
       - airflow.providers.cncf.kubernetes.operators.kubernetes_pod
       - airflow.providers.cncf.kubernetes.operators.pod
       - airflow.providers.cncf.kubernetes.operators.spark_kubernetes
+      - airflow.providers.cncf.kubernetes.operators.resource
 
 sensors:
   - integration-name: Kubernetes
diff --git a/airflow/providers/cncf/kubernetes/utils/delete_from.py b/airflow/providers/cncf/kubernetes/utils/delete_from.py
new file mode 100644
index 0000000000000..59d1896b78f76
--- /dev/null
+++ b/airflow/providers/cncf/kubernetes/utils/delete_from.py
@@ -0,0 +1,157 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+
+# from https://github.com/tomplus/kubernetes_asyncio/pull/239/files
+
+from __future__ import annotations
+
+import re
+
+from kubernetes import client
+from kubernetes.client import ApiClient
+
+DEFAULT_DELETION_BODY = client.V1DeleteOptions(
+    propagation_policy="Background",
+    grace_period_seconds=5,
+)
+
+
+def delete_from_dict(k8s_client, data, body, namespace, verbose=False, **kwargs):
+    api_exceptions = []
+
+    if "List" in data["kind"]:
+        kind = data["kind"].replace("List", "")
+        for yml_doc in data["items"]:
+            if kind != "":
+                yml_doc["apiVersion"] = data["apiVersion"]
+                yml_doc["kind"] = kind
+            try:
+                _delete_from_yaml_single_item(
+                    k8s_client=k8s_client,
+                    yml_document=yml_doc,
+                    verbose=verbose,
+                    namespace=namespace,
+                    body=body,
+                    **kwargs,
+                )
+            except client.rest.ApiException as api_exception:
+                api_exceptions.append(api_exception)
+    else:
+
+        try:
+            _delete_from_yaml_single_item(
+                k8s_client=k8s_client,
+                yml_document=data,
+                verbose=verbose,
+                namespace=namespace,
+                body=body,
+                **kwargs,
+            )
+        except client.rest.ApiException as api_exception:
+            api_exceptions.append(api_exception)
+
+    if api_exceptions:
+        raise FailToDeleteError(api_exceptions)
+
+
+def delete_from_yaml(
+    *,
+    k8s_client: ApiClient,
+    yaml_objects=None,
+    verbose: bool = False,
+    namespace: str = "default",
+    body: dict | None = None,
+    **kwargs,
+):
+    for yml_document in yaml_objects:
+        if yml_document is None:
+            continue
+        else:
+            delete_from_dict(
+                k8s_client=k8s_client,
+                data=yml_document,
+                body=body,
+                namespace=namespace,
+                verbose=verbose,
+                **kwargs,
+            )
+
+
+def _delete_from_yaml_single_item(
+    *,
+    k8s_client: ApiClient,
+    yml_document: dict,
+    verbose: bool = False,
+    namespace: str = "default",
+    body: dict | None = None,
+    **kwargs,
+):
+    if body is None:
+        body = DEFAULT_DELETION_BODY
+
+    # get group and version from apiVersion
+    group, _, version = yml_document["apiVersion"].partition("/")
+    if version == "":
+        version = group
+        group = "core"
+    # Take care for the case e.g. api_type is "apiextensions.k8s.io"
+    # Only replace the last instance
+    group = "".join(group.rsplit(".k8s.io", 1))
+    # convert group name from DNS subdomain format to
+    # python class name convention
+    group = "".join(word.capitalize() for word in group.split("."))
+    fcn_to_call = f"{group}{version.capitalize()}Api"
+    k8s_api = getattr(client, fcn_to_call)(k8s_client)
+    # Replace CamelCased action_type into snake_case
+    kind = yml_document["kind"]
+    kind = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", kind)
+    kind = re.sub("([a-z0-9])([A-Z])", r"\1_\2", kind).lower()
+
+    # Decide which namespace we are going to use for deleting the object
+    # IMPORTANT: the docs namespace takes precedence over the namespace in args
+    #    create_from_yaml_single_item have same behaviour
+    if "namespace" in yml_document["metadata"]:
+        namespace = yml_document["metadata"]["namespace"]
+    name = yml_document["metadata"]["name"]
+
+    # Expect the user to delete namespaced objects more often
+    resp: client.V1Status
+    if hasattr(k8s_api, f"delete_namespaced_{kind}"):
+        resp = getattr(k8s_api, f"delete_namespaced_{kind}")(
+            name=name, namespace=namespace, body=body, **kwargs
+        )
+    else:
+        resp = getattr(k8s_api, f"delete_{kind}")(name=name, body=body, **kwargs)
+    if verbose:
+        print(f"{kind} deleted. status='{str(resp.status)}'")
+    return resp
+
+
+class FailToDeleteError(Exception):
+    """
+    An exception class for handling error if an error occurred when
+    handling a yaml file during deletion of the resource.
+    """
+
+    def __init__(self, api_exceptions: list):
+        self.api_exceptions = api_exceptions
+
+    def __str__(self):
+        msg = ""
+        for api_exception in self.api_exceptions:
+            msg += f"Error from server ({api_exception.reason}):{api_exception.body}\n"
+        return msg
diff --git a/tests/always/test_project_structure.py b/tests/always/test_project_structure.py
index 9f0d36b4e2d61..b17e285bdb33a 100644
--- a/tests/always/test_project_structure.py
+++ b/tests/always/test_project_structure.py
@@ -428,6 +428,7 @@ class TestCncfProviderProjectStructure(ExampleCoverageTest):
         "airflow.providers.cncf.kubernetes.operators.kubernetes_pod",
         "airflow.providers.cncf.kubernetes.triggers.kubernetes_pod",
     }
+    BASE_CLASSES = {"airflow.providers.cncf.kubernetes.operators.resource.KubernetesResourceBaseOperator"}
 
 
 class TestSlackProviderProjectStructure(ExampleCoverageTest):
diff --git a/tests/providers/cncf/kubernetes/operators/test_resource.py b/tests/providers/cncf/kubernetes/operators/test_resource.py
new file mode 100644
index 0000000000000..a565e84fc1aa9
--- /dev/null
+++ b/tests/providers/cncf/kubernetes/operators/test_resource.py
@@ -0,0 +1,120 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+import yaml
+
+from airflow import DAG
+from airflow.providers.cncf.kubernetes.operators.resource import (
+    KubernetesCreateResourceOperator,
+    KubernetesDeleteResourceOperator,
+)
+from airflow.utils import timezone
+
+TEST_VALID_RESOURCE_YAML = """
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: test_pvc
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: standard
+  resources:
+    requests:
+      storage: 5Gi
+"""
+
+TEST_VALID_LIST_RESOURCE_YAML = """
+apiVersion: v1
+kind: List
+items:
+- apiVersion: v1
+  kind: PersistentVolumeClaim
+  metadata:
+    name: test_pvc_1
+- apiVersion: v1
+  kind: PersistentVolumeClaim
+  metadata:
+    name: test_pvc_2
+"""
+
+HOOK_CLASS = "airflow.providers.cncf.kubernetes.hooks.kubernetes.KubernetesHook"
+
+
+@patch("airflow.utils.context.Context")
+class TestKubernetesXResourceOperator:
+    @pytest.fixture(autouse=True)
+    def setup_tests(self, dag_maker):
+        self._default_client_patch = patch(f"{HOOK_CLASS}._get_default_client")
+        self._default_client_mock = self._default_client_patch.start()
+
+        yield
+
+        patch.stopall()
+
+    def setup_method(self):
+        args = {"owner": "airflow", "start_date": timezone.datetime(2020, 2, 1)}
+        self.dag = DAG("test_dag_id", default_args=args)
+
+    @patch("kubernetes.client.api.CoreV1Api.create_namespaced_persistent_volume_claim")
+    def test_create_application_from_yaml(self, mock_create_namespaced_persistent_volume_claim, context):
+        op = KubernetesCreateResourceOperator(
+            yaml_conf=TEST_VALID_RESOURCE_YAML,
+            dag=self.dag,
+            kubernetes_conn_id="kubernetes_default",
+            task_id="test_task_id",
+        )
+
+        op.execute(context)
+
+        mock_create_namespaced_persistent_volume_claim.assert_called_once_with(
+            body=yaml.safe_load(TEST_VALID_RESOURCE_YAML), namespace="default"
+        )
+
+    @patch("kubernetes.client.api.CoreV1Api.delete_namespaced_persistent_volume_claim")
+    def test_single_delete_application_from_yaml(
+        self, mock_delete_namespaced_persistent_volume_claim, context
+    ):
+        op = KubernetesDeleteResourceOperator(
+            yaml_conf=TEST_VALID_RESOURCE_YAML,
+            dag=self.dag,
+            kubernetes_conn_id="kubernetes_default",
+            task_id="test_task_id",
+        )
+
+        op.execute(context)
+
+        mock_delete_namespaced_persistent_volume_claim.assert_called()
+
+    @patch("kubernetes.client.api.CoreV1Api.delete_namespaced_persistent_volume_claim")
+    def test_multi_delete_application_from_yaml(
+        self, mock_delete_namespaced_persistent_volume_claim, context
+    ):
+        op = KubernetesDeleteResourceOperator(
+            yaml_conf=TEST_VALID_LIST_RESOURCE_YAML,
+            dag=self.dag,
+            kubernetes_conn_id="kubernetes_default",
+            task_id="test_task_id",
+        )
+
+        op.execute(context)
+
+        mock_delete_namespaced_persistent_volume_claim.assert_called()
diff --git a/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py b/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
new file mode 100644
index 0000000000000..7be0a8099106b
--- /dev/null
+++ b/tests/system/providers/cncf/kubernetes/example_kubernetes_resource.py
@@ -0,0 +1,83 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+"""
+This is an example DAG which uses KubernetesCreateResourceOperator and KubernetesDeleteResourceOperator.
+In this example, we create two tasks which execute sequentially.
+The first task is to create a PVC on Kubernetes cluster.
+and the second task is to delete the PVC.
+"""
+from __future__ import annotations
+
+import os
+from datetime import datetime, timedelta
+
+from airflow import DAG
+from airflow.providers.cncf.kubernetes.operators.resource import (
+    KubernetesCreateResourceOperator,
+    KubernetesDeleteResourceOperator,
+)
+
+pvc_name = "toto"
+
+pvc_conf = f"""
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: {pvc_name}
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: standard
+  resources:
+    requests:
+      storage: 5Gi
+"""
+
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
+DAG_ID = "example_kubernetes_resource_operator"
+
+with DAG(
+    DAG_ID,
+    default_args={"max_active_runs": 1},
+    description="create and delete a PVC in a kubernetes",
+    schedule=timedelta(days=1),
+    start_date=datetime(2021, 1, 1),
+    catchup=False,
+) as dag:
+    t1 = KubernetesCreateResourceOperator(
+        task_id="create_pvc",
+        yaml_conf=pvc_conf,
+    )
+
+    t2 = KubernetesDeleteResourceOperator(
+        task_id="delete_pvc",
+        yaml_conf=pvc_conf,
+    )
+
+    t1 >> t2
+
+    from tests.system.utils.watcher import watcher
+
+    # This test needs watcher in order to properly mark success/failure
+    # when "tearDown" task with trigger rule is part of the DAG
+    list(dag.tasks) >> watcher()
+
+from tests.system.utils import get_test_run  # noqa: E402
+
+# Needed to run the example DAG with pytest (see: tests/system/README.md#run_via_pytest)
+test_run = get_test_run(dag)
