sdk/python/jobs/spark/submit_spark_standalone_jobs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Submit standalone Spark jobs in Azure Machine Learning (preview)\n",
    "Please see documentation page: [Submit Spark jobs in Azure Machine Learning (preview)](https://learn.microsoft.com/azure/machine-learning/how-to-submit-spark-jobs#submit-a-standalone-spark-job) for more details related to the code samples in this notebook."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Use an attached Synapse Spark pool"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You should have an attached Synapse Spark pool available in your workspace. Please see documentation page: [Attach and manage a Synapse Spark pool in Azure Machine Learning (preview)](https://learn.microsoft.com/azure/machine-learning/how-to-manage-synapse-spark-pool) for more details.\n",
    "\n",
    "**Note** - To ensure successful execution of Spark job, the identity being used for the Spark job should be assigned **Contributor** and **Storage Blob Data Contributor** roles on the Azure storage account used for data input and output."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit a standalone job using managed identity\n",
    "For an attached Synpase Spark pool, managed identity is compute identity of the attached Synapse Spark pool."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azure.ai.ml import MLClient, spark, Input, Output\n",
    "from azure.identity import DefaultAzureCredential\n",
    "from azure.ai.ml.entities import ManagedIdentityConfiguration\n",
    "\n",
    "subscription_id = \"<SUBSCRIPTION_ID>\"\n",
    "resource_group = \"<RESOURCE_GROUP>\"\n",
    "workspace = \"<AML_WORKSPACE_NAME>\"\n",
    "ml_client = MLClient(\n",
    "    DefaultAzureCredential(), subscription_id, resource_group, workspace\n",
    ")\n",
    "\n",
    "spark_job = spark(\n",
    "    display_name=\"Titanic-Spark-Job-SDK-1\",\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",
    "    compute=\"<ATTACHED_SPARK_POOL_NAME>\",\n",
    "    inputs={\n",
    "        \"titanic_data\": Input(\n",
    "            type=\"uri_file\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/titanic.csv\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    outputs={\n",
    "        \"wrangled_data\": Output(\n",
    "            type=\"uri_folder\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/wrangled/\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    identity=ManagedIdentityConfiguration(),\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",
    "\n",
    "# Wait until the job completes\n",
    "ml_client.jobs.stream(returned_spark_job.name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit a standalone job using user identity"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azure.ai.ml import MLClient, spark, Input, Output\n",
    "from azure.identity import DefaultAzureCredential\n",
    "from azure.ai.ml.entities import UserIdentityConfiguration\n",
    "\n",
    "subscription_id = \"<SUBSCRIPTION_ID>\"\n",
    "resource_group = \"<RESOURCE_GROUP>\"\n",
    "workspace = \"<AML_WORKSPACE_NAME>\"\n",
    "ml_client = MLClient(\n",
    "    DefaultAzureCredential(), subscription_id, resource_group, workspace\n",
    ")\n",
    "\n",
    "spark_job = spark(\n",
    "    display_name=\"Titanic-Spark-Job-SDK-2\",\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",
    "    compute=\"<ATTACHED_SPARK_POOL_NAME>\",\n",
    "    inputs={\n",
    "        \"titanic_data\": Input(\n",
    "            type=\"uri_file\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/titanic.csv\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    outputs={\n",
    "        \"wrangled_data\": Output(\n",
    "            type=\"uri_folder\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/wrangled/\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    identity=UserIdentityConfiguration(),\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",
    "\n",
    "# Wait until the job completes\n",
    "ml_client.jobs.stream(returned_spark_job.name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit a standalone job using default identity\n",
    "Default identity for an attached Synpase Spark pool is managed identity (compute identity of the attached Synapse Spark pool)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azure.ai.ml import MLClient, spark, Input, Output\n",
    "from azure.identity import DefaultAzureCredential\n",
    "\n",
    "subscription_id = \"<SUBSCRIPTION_ID>\"\n",
    "resource_group = \"<RESOURCE_GROUP>\"\n",
    "workspace = \"<AML_WORKSPACE_NAME>\"\n",
    "ml_client = MLClient(\n",
    "    DefaultAzureCredential(), subscription_id, resource_group, workspace\n",
    ")\n",
    "\n",
    "spark_job = spark(\n",
    "    display_name=\"Titanic-Spark-Job-SDK-3\",\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",
    "    compute=\"<ATTACHED_SPARK_POOL_NAME>\",\n",
    "    inputs={\n",
    "        \"titanic_data\": Input(\n",
    "            type=\"uri_file\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/titanic.csv\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    outputs={\n",
    "        \"wrangled_data\": Output(\n",
    "            type=\"uri_folder\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/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",
    "\n",
    "# Wait until the job completes\n",
    "ml_client.jobs.stream(returned_spark_job.name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Use a Managed (Automatic) Spark compute"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You should have a Managed (Automatic) Spark compute available in your workspace. Please see [this documentation page](https://learn.microsoft.com/azure/machine-learning/interactive-data-wrangling-with-apache-spark-azure-ml#create-and-configure-managed-automatic-spark-compute-in-azure-machine-learning-notebooks) for more information about creating Managed (Automatic) Spark compute.\n",
    "\n",
    "**Note** - To ensure successful execution of spark job, the identity being used for the Spark job should be assigned **Contributor** and **Storage Blob Data Contributor** roles on the Azure storage account used for data input and output."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit a standalone job using managed identity\n",
    "For a Managed (Automatic) Spark compute, managed identity is the user assigned managed identity attached to the Azure Machine Learning workspace. Please see [this documentation page to learn how to attach the user assigned managed identity](https://learn.microsoft.com/azure/machine-learning/how-to-submit-spark-jobs#ensuring-resource-access-for-spark-jobs) to the Azure Machine Learning workspace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azure.ai.ml import MLClient, spark, Input, Output\n",
    "from azure.identity import DefaultAzureCredential\n",
    "from azure.ai.ml.entities import ManagedIdentityConfiguration\n",
    "\n",
    "subscription_id = \"<SUBSCRIPTION_ID>\"\n",
    "resource_group = \"<RESOURCE_GROUP>\"\n",
    "workspace = \"<AML_WORKSPACE_NAME>\"\n",
    "ml_client = MLClient(\n",
    "    DefaultAzureCredential(), subscription_id, resource_group, workspace\n",
    ")\n",
    "\n",
    "spark_job = spark(\n",
    "    display_name=\"Titanic-Spark-Job-SDK-4\",\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.2.0\",\n",
    "    },\n",
    "    inputs={\n",
    "        \"titanic_data\": Input(\n",
    "            type=\"uri_file\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/titanic.csv\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    outputs={\n",
    "        \"wrangled_data\": Output(\n",
    "            type=\"uri_folder\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/wrangled/\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    identity=ManagedIdentityConfiguration(),\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",
    "\n",
    "# Wait until the job completes\n",
    "ml_client.jobs.stream(returned_spark_job.name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit a standalone job using user identity"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azure.ai.ml import MLClient, spark, Input, Output\n",
    "from azure.identity import DefaultAzureCredential\n",
    "from azure.ai.ml.entities import UserIdentityConfiguration\n",
    "\n",
    "subscription_id = \"<SUBSCRIPTION_ID>\"\n",
    "resource_group = \"<RESOURCE_GROUP>\"\n",
    "workspace_name = \"<AML_WORKSPACE_NAME>\"\n",
    "ml_client = MLClient(\n",
    "    DefaultAzureCredential(), subscription_id, resource_group, workspace_name\n",
    ")\n",
    "\n",
    "spark_job = spark(\n",
    "    display_name=\"Titanic-Spark-Job-SDK-5\",\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.2.0\",\n",
    "    },\n",
    "    inputs={\n",
    "        \"titanic_data\": Input(\n",
    "            type=\"uri_file\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/titanic.csv\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    outputs={\n",
    "        \"wrangled_data\": Output(\n",
    "            type=\"uri_folder\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/wrangled/\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    identity=UserIdentityConfiguration(),\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",
    "\n",
    "# Wait until the job completes\n",
    "ml_client.jobs.stream(returned_spark_job.name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit a standalone job using default identity\n",
    "Default identity for the Managed (Automatic) Spark compute is user identity."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azure.ai.ml import MLClient, spark, Input, Output\n",
    "from azure.identity import DefaultAzureCredential\n",
    "\n",
    "subscription_id = \"<SUBSCRIPTION_ID>\"\n",
    "resource_group = \"<RESOURCE_GROUP>\"\n",
    "workspace = \"<AML_WORKSPACE_NAME>\"\n",
    "ml_client = MLClient(\n",
    "    DefaultAzureCredential(), subscription_id, resource_group, workspace\n",
    ")\n",
    "\n",
    "spark_job = spark(\n",
    "    display_name=\"Titanic-Spark-Job-SDK-6\",\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.2.0\",\n",
    "    },\n",
    "    inputs={\n",
    "        \"titanic_data\": Input(\n",
    "            type=\"uri_file\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/data/titanic.csv\",\n",
    "            mode=\"direct\",\n",
    "        ),\n",
    "    },\n",
    "    outputs={\n",
    "        \"wrangled_data\": Output(\n",
    "            type=\"uri_folder\",\n",
    "            path=\"azureml://datastores/workspaceblobstore/paths/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",
    "\n",
    "# Wait until the job completes\n",
    "ml_client.jobs.stream(returned_spark_job.name)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.10 - SDK V2",
   "language": "python",
   "name": "python310-sdkv2"
  },
  "language_info": {
   "name": "python",
   "version": "3.7.10"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "6aeff17a1aa7735c2f7cb3a6d691fe1b4d4c3b8d2d650f644ad0f24e1b8e3f3f"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}