sdk/python/jobs/spark/submit_spark_standalone_jobs_managed_vnet.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "# Azure Machine Learning Serverless Spark with Managed Virtual Network\n",
    "This Notebook provides sample codes for running a Spark job using [Azure Machine Learning serverless Spark compute with a managed virtual network](https://learn.microsoft.com/en-us/azure/machine-learning/how-to-managed-network#configure-for-serverless-spark-jobs). In this sample notebook you will:\n",
    "- Create an Azure Machine Learning Workspace with Public Network Access _disabled_.\n",
    "- Configure outbound rules for the Azure Machine Learning workspace that allow storage account data access.\n",
    "- Provision managed network for the workspace.\n",
    "- View created outbound rules.\n",
    "- Submit a Spark job using serverless Spark compute."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Getting MLClient instance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684530155377
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "# import required libraries\n",
    "from azure.ai.ml import MLClient\n",
    "from azure.identity import DefaultAzureCredential\n",
    "\n",
    "# Enter the details of your subscription\n",
    "subscription_id = \"<SUBSCRIPTION_ID>\"\n",
    "resource_group = \"<RESOURCE_GROUP>\"\n",
    "\n",
    "# get an instance of MLClient\n",
    "ml_client = MLClient(DefaultAzureCredential(), subscription_id, resource_group)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Create a Workspace\n",
    "Define a managed VNet with isolation mode `IsolationMode.ALLOW_INTERNET_OUTBOUND` and a user-defined outbound rule for Azure Blob storage account. In this example, Public Network Access to the workspace is _disabled_. The code in cell creates the workspace, but the managed VNet and Private Endpoints corresponding to the outbound rules are provisioned in the later step.\n",
    "\n",
    "> [!NOTE]\n",
    "> If you want to allow only approved outbound traffic to enable data exfiltration protection (DEP), use `IsolationMode.ALLOW_ONLY_APPROVED_OUTBOUND`.\n",
    "\n",
    "If the Azure Blob storage account needs to have Public Network Access _disabled_, then access should be disabled before adding the outbound rule and provisioning the managed VNet for the workspace. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684530258638
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "# Creating a workspace with unique name\n",
    "from azure.ai.ml.entities import Workspace, ManagedNetwork, PrivateEndpointDestination\n",
    "from azure.ai.ml.constants._workspace import IsolationMode\n",
    "\n",
    "# Enter workspace name and the region where the\n",
    "# workspace will be created\n",
    "ws_name = \"<AML_WORKSPACE_NAME>\"\n",
    "region = \"<AZURE_REGION_NAME>\"\n",
    "# Enter Azure Blob storage account name for the outbound rule\n",
    "blob_storage_account = \"<STORAGE_ACCOUNT_NAME>\"\n",
    "\n",
    "ws_mvnet = Workspace(\n",
    "    name=ws_name,\n",
    "    location=region,\n",
    "    hbi_workspace=False,\n",
    "    public_network_access=\"Disabled\",  # Comment this out to enable Public Network Access\n",
    "    tags=dict(purpose=\"demo\"),\n",
    ")\n",
    "\n",
    "ws_mvnet.managed_network = ManagedNetwork(\n",
    "    isolation_mode=IsolationMode.ALLOW_INTERNET_OUTBOUND\n",
    ")\n",
    "\n",
    "rule_name = \"<OUTBOUND_RULE_NAME>\"\n",
    "service_resource_id = f\"/subscriptions/{subscription_id}/resourceGroups/{resource_group}/providers/Microsoft.Storage/storageAccounts/{blob_storage_account}\"\n",
    "subresource_target = \"blob\"\n",
    "spark_enabled = True\n",
    "\n",
    "ws_mvnet.managed_network.outbound_rules = [\n",
    "    PrivateEndpointDestination(\n",
    "        name=rule_name,\n",
    "        service_resource_id=service_resource_id,\n",
    "        subresource_target=subresource_target,\n",
    "        spark_enabled=spark_enabled,\n",
    "    )\n",
    "]\n",
    "\n",
    "print(f\"Initiating creation request for workspace with name: {ws_name}\")\n",
    "ws_mvnet = ml_client.workspaces.begin_create(ws_mvnet).result()\n",
    "print(\"workspace created!\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Provision the Managed VNet\n",
    "Provision the managed VNet for the workspace that we created in the previous step. The method `begin_provision_network`:\n",
    "- Provisions a managed VNet for the workspace.\n",
    "- Creates Private Endpoints defined by outbound rules.\n",
    "- Creates system-defined Private Endpoints.\n",
    "- Enables Spark support based on the passed parameter value.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684435217796
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "# Provisioning managed VNet with Spark support\n",
    "include_spark = True\n",
    "provision_network_result = ml_client.workspaces.begin_provision_network(\n",
    "    workspace_name=ws_name, include_spark=include_spark\n",
    ").result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Get workspace details\n",
    "Showing workspace details."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684435370053
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "print(f\"Getting workspace with name: {ws_name}\")\n",
    "ws_mvnet = ml_client.workspaces.get(ws_name)\n",
    "print(ws_mvnet)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### List outbound rules for the workspace\n",
    "Listing outbound rules for the workspace. This list shows Private Endpoints created for defined outbound rules, and system-defined Private Endpoints."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684524544810
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "# List outbound rules for a workspace\n",
    "rule_list = ml_client._workspace_outbound_rules.list(ws_name)\n",
    "print([r._to_dict() for r in rule_list])"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Display details of an outbound rule\n",
    "Displaying details of an outbound rule."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684435395108
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "# Get details of an outbound rule by name\n",
    "rule = ml_client._workspace_outbound_rules.get(ws_name, rule_name)\n",
    "print(rule)\n",
    "print(rule._to_dict())"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Submit a Spark job\n",
    "In the subsequent parts:\n",
    "- A directory named `src` is created to keep Python scripts used in a standalone Spark job.\n",
    "- A Python script to wrangle Titanic data is written to the `src` directory.\n",
    "- A standalone job is submitted using the data stored on the Azure Blob storage account and serverless Spark compute with managed VNet."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Ensure code path exists\n",
    "Create a directory named `src` in the current directory, if it does not exist."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684435409121
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "os.makedirs(\"src\", exist_ok=True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "#### Write a script file\n",
    "This script file uses Azure Blob storage access key stored in an Azure Key Vault for credential-based data access. Identity-based data access is not supported for Spark jobs using data in Azure Blob storage accounts.\n",
    "\n",
    "If you want to use a SAS token instead of `access_key`, you can use the following to get a `sas_token` and set it in the configuration:\n",
    "```python\n",
    "sas_token = token_library.getSecret(\"<KEY_VAULT_NAME>\", \"<SAS_TOKEN_SECRET_NAME>\")\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net\",\n",
    "    sas_token,\n",
    ")\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "%%writefile src/titanic.py\n",
    "import argparse\n",
    "from operator import add\n",
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "parser = argparse.ArgumentParser()\n",
    "parser.add_argument(\"--titanic_data\")\n",
    "parser.add_argument(\"--wrangled_data\")\n",
    "\n",
    "args = parser.parse_args()\n",
    "print(\"Input path: \" + args.wrangled_data)\n",
    "print(\"Output path: \" + args.titanic_data)\n",
    "\n",
    "sc = SparkSession.builder.getOrCreate()\n",
    "token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary\n",
    "access_key = token_library.getSecret(\"<KEY_VAULT_NAME>\", \"<ACCESS_KEY_SECRET_NAME>\")\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net\", access_key\n",
    ")\n",
    "\n",
    "df = pd.read_csv(args.titanic_data, index_col=\"PassengerId\")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(args.wrangled_data, index_col=\"PassengerId\")\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "#### Get MLClient for the workspace with managed VNet\n",
    "Getting instance of `MLClient` for the workspace with managed VNet provisioned."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684518058481
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "ml_client = MLClient(\n",
    "    DefaultAzureCredential(), subscription_id, resource_group, workspace_name=ws_name\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "#### Submit the job\n",
    "Submit a standalone Spark job to wrangle Titanic data stored in an Azure Blob storage account. To learn more about submitting a standalone Spark job, [see this documentation page](https://learn.microsoft.com/azure/machine-learning/how-to-submit-spark-jobs?view=azureml-api-2&tabs=sdk#submit-a-standalone-spark-job)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684519240489
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "from azure.ai.ml import spark, Input, Output\n",
    "\n",
    "# Enter the Azure Blob storage account name and container name.\n",
    "# The file `titanic.csv` should be placed inside folder `data`\n",
    "# created in the Azure Blob storage container.\n",
    "blob_storage_account = \"<STORAGE_ACCOUNT_NAME>\"\n",
    "container_name = \"<BLOB_CONTAINER_NAME>\"\n",
    "\n",
    "spark_job = spark(\n",
    "    display_name=\"Job from serverless Spark with VNet using Azure Blob storage\",\n",
    "    code=\"./src\",\n",
    "    entry={\"file\": \"titanic.py\"},\n",
    "    driver_cores=1,\n",
    "    driver_memory=\"2g\",\n",
    "    executor_cores=2,\n",
    "    executor_memory=\"2g\",\n",
    "    executor_instances=2,\n",
    "    resources={\n",
    "        \"instance_type\": \"Standard_E8S_V3\",\n",
    "        \"runtime_version\": \"3.3.0\",\n",
    "    },\n",
    "    inputs={\n",
    "        \"titanic_data\": Input(\n",
    "            type=\"uri_file\",\n",
    "            path=f\"wasbs://{container_name}@{blob_storage_account}.blob.core.windows.net/data/titanic.csv\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    outputs={\n",
    "        \"wrangled_data\": Output(\n",
    "            type=\"uri_folder\",\n",
    "            path=f\"wasbs://{container_name}@{blob_storage_account}.blob.core.windows.net/data/wrangled\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    args=\"--titanic_data ${{inputs.titanic_data}} --wrangled_data ${{outputs.wrangled_data}}\",\n",
    ")\n",
    "\n",
    "returned_spark_job = ml_client.jobs.create_or_update(spark_job)\n",
    "print(returned_spark_job.id)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Add another outbound rule\n",
    "Adding another outbound rule and creating a Private Endpoint to an Azure Data Lake Storage (ADLS) Gen2 account. A separate call to provision Private Endpoint is _not_ required.\n",
    "\n",
    "If the Azure Data Lake Storage (ADLS) Gen2 account needs to have Public Network Access _disabled_, then access should be disabled before adding the outbound rule and updating the workspace to create the Private Endpoint. The [private endpoints for the workspace and the storage accounts should be in the same VNet](https://learn.microsoft.com/azure/machine-learning/how-to-secure-workspace-vnet#limitations).\n",
    "\n",
    "Note that `subresource_target` value `dfs` is used here for the Azure Data Lake Storage (ADLS) Gen2 account."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684514387513
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "from azure.ai.ml.entities import PrivateEndpointDestination\n",
    "\n",
    "# This will add a new outbound rule to existing rules\n",
    "rule_name = \"<OUTBOUND_RULE_NAME_GEN2>\"  # This name should be unique\n",
    "adls_storage_account = \"<GEN2_STORAGE_ACCOUNT_NAME>\"\n",
    "service_resource_id = f\"/subscriptions/{subscription_id}/resourceGroups/{resource_group}/providers/Microsoft.Storage/storageAccounts/{adls_storage_account}\"\n",
    "subresource_target = \"dfs\"\n",
    "spark_enabled = True\n",
    "\n",
    "ws = ml_client.workspaces.get()\n",
    "\n",
    "ws.managed_network.outbound_rules = [\n",
    "    PrivateEndpointDestination(\n",
    "        name=rule_name,\n",
    "        service_resource_id=service_resource_id,\n",
    "        subresource_target=subresource_target,\n",
    "        spark_enabled=spark_enabled,\n",
    "    )\n",
    "]\n",
    "\n",
    "ml_client.workspaces.begin_update(ws).result()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### List outbound rules for the workspace\n",
    "Listing outbound rules for the workspace. This list shows Private Endpoints created for defined outbound rules, and system-defined Private Endpoints."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684527041002
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "# List outbound rules for a workspace\n",
    "rule_list = ml_client._workspace_outbound_rules.list(ws_name)\n",
    "print([r._to_dict() for r in rule_list])"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Write Python script for a job with ADLS Gen2\n",
    "This script file assumes data stored in an Azure Data Lake Storage (ADLS) Gen2 accessed using identity-based data access. The user identity should have **Contributor** and **Storage Blob Data Contributor** roles assigned to ensure data access."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "%%writefile src/titanic-adlsg2.py\n",
    "import argparse\n",
    "from operator import add\n",
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "parser = argparse.ArgumentParser()\n",
    "parser.add_argument(\"--titanic_data\")\n",
    "parser.add_argument(\"--wrangled_data\")\n",
    "\n",
    "args = parser.parse_args()\n",
    "print(\"Input path: \" + args.wrangled_data)\n",
    "print(\"Output path: \" + args.titanic_data)\n",
    "\n",
    "df = pd.read_csv(args.titanic_data, index_col=\"PassengerId\")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(args.wrangled_data, index_col=\"PassengerId\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Submit the job\n",
    "Submit a standalone Spark job to wrangle Titanic data stored in an Azure Data Lake Storage (ADLS) Gen2 account. To learn more about submitting a standalone Spark job, [see this documentation page](https://learn.microsoft.com/azure/machine-learning/how-to-submit-spark-jobs?view=azureml-api-2&tabs=sdk#submit-a-standalone-spark-job)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1684515565112
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "from azure.ai.ml import spark, Input, Output\n",
    "from azure.ai.ml.entities import UserIdentityConfiguration\n",
    "\n",
    "# Enter the Azure Data Lake Storage (ADLS) Gen2 account name and container name.\n",
    "# The file `titanic.csv` should be placed inside folder `data`\n",
    "# created in the Azure Data Lake Storage (ADLS) Gen2 container.\n",
    "adls_storage_account = \"<GEN2_STORAGE_ACCOUNT_NAME>\"\n",
    "container_name = \"<ADLS_CONTAINER_NAME>\"\n",
    "\n",
    "spark_job = spark(\n",
    "    display_name=\"Job from serverless Spark with VNet using data in ADLS Gen2\",\n",
    "    code=\"./src\",\n",
    "    entry={\"file\": \"titanic-adlsg2.py\"},\n",
    "    driver_cores=1,\n",
    "    driver_memory=\"2g\",\n",
    "    executor_cores=2,\n",
    "    executor_memory=\"2g\",\n",
    "    executor_instances=2,\n",
    "    resources={\n",
    "        \"instance_type\": \"Standard_E8S_V3\",\n",
    "        \"runtime_version\": \"3.3.0\",\n",
    "    },\n",
    "    inputs={\n",
    "        \"titanic_data\": Input(\n",
    "            type=\"uri_file\",\n",
    "            path=f\"abfss://{container_name}@{adls_storage_account}.dfs.core.windows.net/data/titanic.csv\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    outputs={\n",
    "        \"wrangled_data\": Output(\n",
    "            type=\"uri_folder\",\n",
    "            path=f\"abfss://{container_name}@{adls_storage_account}.dfs.core.windows.net/data/wrangled\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    args=\"--titanic_data ${{inputs.titanic_data}} --wrangled_data ${{outputs.wrangled_data}}\",\n",
    "    identity=UserIdentityConfiguration(),\n",
    ")\n",
    "\n",
    "returned_spark_job = ml_client.jobs.create_or_update(spark_job)\n",
    "print(returned_spark_job.id)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Delete an outbound rule"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1683671276584
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "# Delete an outbound rule from a workspace\n",
    "ml_client._workspace_outbound_rules.begin_remove(ws_name, rule_name).result()\n",
    "\n",
    "# List outbound rules for the workspace\n",
    "rule_list = ml_client._workspace_outbound_rules.list(ws_name)\n",
    "print([r._to_dict() for r in rule_list])"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Clean-up\n",
    "Deleting the workspace also deletes corresponding managed VNet and Private Endpoints."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "gather": {
     "logged": 1683671365527
    },
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "ml_client.workspaces.begin_delete(\n",
    "    name=ws_name, permanently_delete=True, delete_dependent_resources=True\n",
    ")"
   ]
  }
 ],
 "metadata": {
  "kernel_info": {
   "name": "azureml_py310_sdkv2_vnet"
  },
  "kernelspec": {
   "display_name": "azureml_py310_sdkv2_vnet",
   "language": "python",
   "name": "azureml_py310_sdkv2_vnet"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.11"
  },
  "microsoft": {
   "host": {
    "AzureML": {
     "notebookHasBeenCompleted": true
    }
   },
   "ms_spell_check": {
    "ms_spell_check_language": "en"
   }
  },
  "nteract": {
   "version": "nteract-front-end@1.0.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}