job_metadata_information.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "b937bc48-f902-44b0-bd0a-4f0f6867fcc0",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "####Installed generic libraries "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "f94d6911-b3e1-4069-a74f-9b033195fb8e",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "###########################################################################################################\n",
    "##                 Generic Notebook parameters    - Here we are check workspace enevironment and based on that we are setting the path             \n",
    "###########################################################################################################\n",
    "if spark.conf.get(\"spark.databricks.clusterUsageTags.managedResourceGroup\") == 'rg-training-team-dev-01':\n",
    "    environment = \"dev\"\n",
    "    print('This is', environment, \"environment\")\n",
    "else:\n",
    "    environment = \"prod\"  # Empty means we can consider as Production environment. Created for email delivery\n",
    "    print('This is', environment, \"environment\")\n",
    "\n",
    "# delta_path = getStoragePath(spark.conf.get(\"spark.databricks.clusterUsageTags.managedResourceGroup\"), \"silver\")\n",
    "ConfigJobmetadata = f'abfss://config@dlstraining{environment}01.dfs.core.windows.net/config/job_metadata_details/delta'\n",
    "print(ConfigJobmetadata)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Generate Personal access token and Access from keyvault for databriks API connectivity"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "9bdc23a7-f560-49f7-89ee-59e2040bd613",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "instance_name = spark.conf.get(\"spark.databricks.workspaceUrl\")\n",
    "workspace_url = f\"https://{instance_name}\"\n",
    "workspace_name = f\"dbs-training-{environment}-01\"\n",
    "api_token = dbutils.secrets.get(scope = f\"kv-training-01\", key = f\"dbs-training-{environment}-01-pat-token\")\n",
    "print(\"Databricks Instance URL:\", workspace_url,\"Databricks Instance Name:\",workspace_name )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "e50647ee-e2d6-4f75-8426-ead4f92fe2e6",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "source": [
    "This notebook will extract all job configuration in central table using databricks api so we can see all job details in dashboard and we can easily take action. \n",
    "     Github confoguration \n",
    "     Cluster configration\n",
    "     Execution schedule "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "2e692487-71cd-43fb-85fa-b2961215abbc",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "\n",
    "import requests\n",
    "import pandas as pd\n",
    "from datetime import datetime,timedelta\n",
    "\n",
    "def fetch_job_information(api_token, workspace_url):\n",
    "    headers = {\n",
    "        \"Authorization\": f\"Bearer {api_token}\",\n",
    "        \"Content-Type\": \"application/json\"\n",
    "    }\n",
    "    # Endpoint to list all jobs\n",
    "    jobs_endpoint = f\"{workspace_url}/api/2.0/jobs/list\"\n",
    "    \n",
    "    # Sending GET request to fetch job information\n",
    "    response = requests.get(jobs_endpoint, headers=headers)\n",
    "    response_json = response.json()\n",
    "    \n",
    "    # Extracting relevant job information\n",
    "    job_data = []\n",
    "    for job in response_json[\"jobs\"]:\n",
    "        #print(job)\n",
    "        job_id = job[\"job_id\"]\n",
    "        job_name = job[\"settings\"][\"name\"]\n",
    "        try:\n",
    "            creator_name = job[\"creator_user_name\"]\n",
    "        except KeyError:\n",
    "            creator_name = \"Unknown\"\n",
    "        cluster_node_type_id = job[\"settings\"].get(\"new_cluster\", {}).get(\"node_type_id\", \"Unknown\")\n",
    "        cluster_spark_version = job[\"settings\"].get(\"new_cluster\", {}).get(\"spark_version\", \"Unknown\")\n",
    "        cluster_runtime_engine = job[\"settings\"].get(\"new_cluster\", {}).get(\"runtime_engine\", \"Unknown\")\n",
    "        cluster_num_workers = job[\"settings\"].get(\"new_cluster\", {}).get(\"num_workers\", \"Unknown\")\n",
    "        data_security_mode = job[\"settings\"].get(\"new_cluster\", {}).get(\"data_security_mode\", \"Unknown\")\n",
    "        job_created_time = datetime.fromtimestamp(job[\"created_time\"]/ 1000)\n",
    "        cron_expression = job[\"settings\"].get(\"schedule\", {}).get(\"quartz_cron_expression\", \"Unknown\") \n",
    "        cron_timezone = job[\"settings\"].get(\"schedule\", {}).get(\"timezone_id\", \"Unknown\")\n",
    "        pause_status = job[\"settings\"].get(\"schedule\", {}).get(\"pause_status\", \"Unknown\")\n",
    "       \n",
    "        # Appending job information to the list\n",
    "        job_data.append((job_id, job_name,creator_name,cluster_node_type_id,cluster_spark_version,cluster_runtime_engine,cluster_num_workers,data_security_mode,job_created_time,cron_expression,cron_timezone,pause_status))\n",
    "        \n",
    "    # Creating a DataFrame from the extracted job information\n",
    "    columns = [\"job_id\", \"job_name\",\"creator_name\",\"cluster_node_type_id\",\"cluster_spark_version\",\"cluster_runtime_engine\",\"cluster_num_workers\",\"data_security_mode\",\"job_created_time\",\"cron_expression\",\"cron_timezone\",\"pause_status\"]\n",
    "    job_df = pd.DataFrame(job_data, columns=columns)\n",
    "    \n",
    "    return job_df\n",
    "\n",
    "# Set your Databricks API token and workspace URL\n",
    "api_token = dbutils.secrets.get(scope = f\"kv-training-01\", key = f\"dbs-training-{environment}-01-pat-token\")\n",
    "\n",
    "# Fetch job information and store it in a DataFrame\n",
    "job_information_df = fetch_job_information(api_token, workspace_url)\n",
    "\n",
    "# New columns\n",
    "workspace_name = workspace_name\n",
    "workspace_url = workspace_url\n",
    "\n",
    "# Inserting new columns\n",
    "job_information_df.insert(2, 'workspace_name', workspace_name)\n",
    "job_information_df.insert(3, 'workspace_url', workspace_url)\n",
    "\n",
    "sparkdf = spark.createDataFrame(job_information_df)\n",
    "sparkdf.createOrReplaceTempView(\"v_job_information\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create config_dev/prod catalog for storing all configration details informations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "62aa4b4b-1768-4d62-a17e-dc3f8c319fed",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "InsertQuery = f\"MERGE INTO config_{environment}.config.job_metadata_details AS target \" \\\n",
    "              \"USING v_job_information AS source \" \\\n",
    "              \"ON target.job_id = source.job_id \" \\\n",
    "              \"WHEN MATCHED THEN \" \\\n",
    "              \"  UPDATE SET target.job_name = source.job_name, \" \\\n",
    "              \"             target.cluster_node_type_id = source.cluster_node_type_id, \" \\\n",
    "              \"             target.cluster_spark_version = source.cluster_spark_version, \" \\\n",
    "              \"             target.cluster_runtime_engine = source.cluster_runtime_engine, \" \\\n",
    "              \"             target.cluster_num_workers = source.cluster_num_workers, \" \\\n",
    "              \"             target.data_security_mode = source.data_security_mode, \" \\\n",
    "              \"             target.cron_expression = source.cron_expression, \" \\\n",
    "              \"             target.cron_timezone = source.cron_timezone, \" \\\n",
    "              \"             target.pause_status = source.pause_status \" \\\n",
    "              \"WHEN NOT MATCHED THEN \" \\\n",
    "              \"  INSERT ( \" \\\n",
    "              \"          job_id, \" \\\n",
    "              \"          job_name, \" \\\n",
    "              \"          workspace_name , \" \\\n",
    "              \"          workspace_url , \" \\\n",
    "              \"          creator_name , \" \\\n",
    "              \"          cluster_node_type_id , \" \\\n",
    "              \"          cluster_spark_version , \" \\\n",
    "              \"          cluster_runtime_engine , \" \\\n",
    "              \"          cluster_num_workers , \" \\\n",
    "              \"          data_security_mode , \" \\\n",
    "              \"          job_created_time , \" \\\n",
    "              \"          cron_expression , \" \\\n",
    "              \"          cron_timezone , \" \\\n",
    "              \"          pause_status \" \\\n",
    "              \"        ) \" \\\n",
    "              \"  VALUES ( \" \\\n",
    "              \"            source.job_id, \" \\\n",
    "              \"            source.job_name, \" \\\n",
    "              \"            source.workspace_name , \" \\\n",
    "              \"            source.workspace_url , \" \\\n",
    "              \"            source.creator_name , \" \\\n",
    "              \"            source.cluster_node_type_id , \" \\\n",
    "              \"            source.cluster_spark_version , \" \\\n",
    "              \"            source.cluster_runtime_engine , \" \\\n",
    "              \"            source.cluster_num_workers , \" \\\n",
    "              \"            source.data_security_mode , \" \\\n",
    "              \"            source.job_created_time , \" \\\n",
    "              \"            source.cron_expression , \" \\\n",
    "              \"            source.cron_timezone , \" \\\n",
    "              \"            source.pause_status \" \\\n",
    "              \"          )\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Insert Job information metadata in job_metadata_details table "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "84c4855e-86d8-4174-8748-dad546369270",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "Query = spark.sql(InsertQuery)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "43f6254c-51d9-41b6-9e9e-061966f1ee55",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "print(\"Total Inserted Records\",Query.select('num_inserted_rows').collect())\n",
    "print(\"Total update Records\",Query.select('num_updated_rows').collect())\n",
    " "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 0,
   "metadata": {
    "application/vnd.databricks.v1+cell": {
     "cellMetadata": {},
     "inputWidgets": {},
     "nuid": "107c4978-334c-46ff-978d-6630e1dafb1d",
     "showTitle": false,
     "tableResultSettingsMap": {},
     "title": ""
    }
   },
   "outputs": [],
   "source": [
    "print(\"Succeeded\")"
   ]
  }
 ],
 "metadata": {
  "application/vnd.databricks.v1+notebook": {
   "computePreferences": null,
   "dashboards": [],
   "environmentMetadata": null,
   "language": "python",
   "notebookMetadata": {
    "pythonIndentUnit": 4
   },
   "notebookName": "job_metadata_information",
   "widgets": {}
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}