code: def create_cluster(
      yandex_conn_id: str | None = None,
      folder_id: str | None = None,
      network_id: str | None = None,
      subnet_id: str | None = None,
      zone: str = YC_ZONE_NAME,
      service_account_id: str | None = None,
      ssh_public_key: str | None = None,
      *,
      dag: DAG | None = None,
      ts_nodash: str | None = None,
  ) -> str:
      hook = YandexCloudBaseHook(yandex_conn_id=yandex_conn_id)
      folder_id = folder_id or hook.default_folder_id
      if subnet_id is None:
          network_id = network_id or hook.sdk.helpers.find_network_id(
              folder_id
          )
          subnet_id = hook.sdk.helpers.find_subnet_id(
              folder_id=folder_id, zone_id=zone, network_id=network_id
          )
      service_account_id = (
          service_account_id
          or hook.sdk.helpers.find_service_account_id()
      )
      ssh_public_key = ssh_public_key or hook.default_public_ssh_key
  
      dag_id = dag and dag.dag_id or "dag"
  
      request = create_cluster_request(
          folder_id=folder_id,
          subnet_id=subnet_id,
          zone=zone,
          cluster_name=f"airflow_{dag_id}_{ts_nodash}"[:62],
          cluster_desc="Created via Airflow custom hook task",
          service_account_id=service_account_id,
          ssh_public_key=ssh_public_key,
          resources=common_pb.Resources(
              resource_preset_id="s2.micro",
              disk_type_id="network-ssd",
          ),
      )
      operation = hook.sdk.client(
          cluster_service_grpc_pb.ClusterServiceStub
      ).Create(request)
      operation_result = hook.sdk.wait_operation_and_get_result(
          operation,
          response_type=cluster_pb.Cluster,
          meta_type=cluster_service_pb.CreateClusterMetadata,
      )
      return operation_result.response.id
  
file_path: airflow/tests/system/providers/yandex/example_yandexcloud.py
imports: from __future__ import annotations
  import os
  from datetime import datetime
  import yandex.cloud.dataproc.v1.cluster_pb2
  import yandex.cloud.dataproc.v1.cluster_service_pb2
  import yandex.cloud.dataproc.v1.cluster_service_pb2_grpc
  import yandex.cloud.dataproc.v1.common_pb2
  import yandex.cloud.dataproc.v1.job_pb2
  import yandex.cloud.dataproc.v1.job_service_pb2
  import yandex.cloud.dataproc.v1.job_service_pb2_grpc
  import yandex.cloud.dataproc.v1.subcluster_pb2
  from google.protobuf.json_format import MessageToDict
  from airflow import DAG
  from airflow.decorators import task
  from airflow.providers.yandex.hooks.yandex import YandexCloudBaseHook
  from tests.system.utils import get_test_run
label_desc: Use image node_id 2 for calling a global function with example usage: create_cluster(yandex_conn_id, folder_id, network_id, subnet_id, zone, service_account_id, ssh_public_key) and returns: operation_result