notebook_examples/pyspark-data_catalog-hive_metastore-data_flow.ipynb

{
 "cells": [
  {
   "cell_type": "raw",
   "id": "c1874fce",
   "metadata": {},
   "source": [
    "@notebook{pyspark-data_catalog-hive_metastore-data_flow.ipynb,\n",
    "    title: Using Data Catalog Metastore with DataFlow,\n",
    "    summary: Write and test a Data Flow batch application using the Oracle Cloud Infrastructure (OCI) Data Catalog Metastore. Configure the job, run the application and clean up resources.,\n",
    "    developed_on: pyspark30_p37_cpu_v5,\n",
    "    keywords: data catalog metastore, data flow,\n",
    "    license: Universal Permissive License v 1.0\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b6627d58",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Upgrade Oracle ADS to pick up latest features and maintain compatibility with Oracle Cloud Infrastructure.\n",
    "\n",
    "!pip install -U oracle-ads"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ad5fed93",
   "metadata": {},
   "source": [
    "Oracle Data Science service sample notebook.\n",
    "\n",
    "Copyright (c) 2022 Oracle, Inc. All rights reserved. Licensed under the [Universal Permissive License v 1.0](https://oss.oracle.com/licenses/upl).\n",
    "\n",
    "---\n",
    "\n",
    "# <font color=\"red\">Using Data Catalog (Hive) Metastore with DataFlow</font>\n",
    "<p style=\"margin-left:10%; margin-right:10%;\">by the <font color=\"teal\">Oracle Cloud Infrastructure Data Science Service.</font></p>\n",
    "\n",
    "---\n",
    "\n",
    "# Overview:\n",
    "\n",
    "This notebook demonstrates how to write and test a Data Flow batch application using the Oracle Cloud Infrastructure (OCI) Data Catalog Metastore. [Oracle Cloud Infrastructure (OCI) Data Catalog](https://docs.oracle.com/en-us/iaas/data-catalog/home.htm) is a metadata management service that helps data professionals discover data and support data governance.  The [Data Catalog Hive Metastore](https://docs.oracle.com/en-us/iaas/data-catalog/using/metastore.htm) provides schema definitions for objects in structured and unstructured data assets backed by Object Store. [Data Flow](https://docs.oracle.com/en-us/iaas/data-flow/using/home.htm) is a fully managed service for running [Apache Spark](https://spark.apache.org/) applications. You write and test a Data Flow batch application using the Data Catalog Metastore in this notebook.\n",
    "\n",
    "Compatible conda pack: [PySpark 3.0 and Data Flow](https://docs.oracle.com/en-us/iaas/data-science/using/conda-pyspark-fam.htm) for CPU on Python 3.7 (version 5.0)\n",
    "\n",
    "---\n",
    "\n",
    "## Contents:\n",
    "\n",
    " - <a href='#intro'>Introduction</a>\n",
    "     - <a href='#prerequisite'>Setup</a>\n",
    "         - <a href='#policy'>Policy</a>\n",
    "         - <a href='#var'>Variables</a>\n",
    " - <a href='#appscript'>Application Script</a>\n",
    " - <a href='#jobs'>Create and Run a Data Flow Application</a>\n",
    "     - <a href='#conf'>Configurating Job</a>\n",
    "     - <a href='#run'>Run the Data Flow Application</a>\n",
    " - <a href='#clean_up'>Clean Up</a>\n",
    " - <a href='#ref'>References</a>\n",
    "\n",
    "---\n",
    "\n",
    "\n",
    "Datasets are provided as a convenience.  Datasets are considered third-party content and are not considered materials \n",
    "under your agreement with Oracle.\n",
    "\n",
    "You can access the `orcl_attrition` dataset license [here](https://oss.oracle.com/licenses/upl).\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "03e8de59",
   "metadata": {},
   "outputs": [],
   "source": [
    "import ads\n",
    "import os\n",
    "import tempfile\n",
    "import shutil\n",
    "\n",
    "from ads.jobs.ads_job import Job\n",
    "from ads.jobs import DataFlow, DataFlowRun, DataFlowRuntime\n",
    "from uuid import uuid4\n",
    "\n",
    "ads.set_auth(auth=\"resource_principal\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8d2f6e96",
   "metadata": {},
   "source": [
    "<a id='intro'></a>\n",
    "# Introduction \n",
    "\n",
    "[Oracle Cloud Infrastructure (OCI) Data Catalog](https://docs.oracle.com/en-us/iaas/data-catalog/home.htm) is a metadata management service that helps data professionals discover data and support data governance.  The [Data Catalog Hive Metastore](https://docs.oracle.com/en-us/iaas/data-catalog/using/metastore.htm) provides schema definitions for objects in structured and unstructured data assets. The Metastore is the central metadata repository to understand tables backed by files on object storage. [Data Flow](https://docs.oracle.com/en-us/iaas/data-flow/using/home.htm) is a fully managed service for running [Apache Spark](https://spark.apache.org/) applications. [Data Science jobs](https://docs.oracle.com/en-us/iaas/tools/ads-sdk/latest/user_guide/jobs/overview.html) allow you to run customized tasks outside of a notebook session. As a Data Flow user, you can access the Data Catalog Metastore to securely store and retrieve schema definitions for data assets. For integration with Data Flow, the Metastore provides an invocation endpoint to Data Flow. This endpoint exposes the Hive Metastore interface. [Apache Hive](https://hive.apache.org/) is a data warehousing framework that facilitates read, write, or manage operations on large datasets residing in distributed systems. A Hive Metastore is the central repository of metadata for a Hive cluster. It stores metadata for data structures such as databases, tables, and partitions in a relational database, backed by files on Object Storage. Apache Spark SQL makes use of a Hive Metastore for this purpose.\n",
    "\n",
    "<a id='prerequisite'></a>\n",
    "## Setup\n",
    "\n",
    "<a id='policy'></a>\n",
    "### Policy\n",
    "\n",
    "To control who has access to Data Flow, and the type of access for each group of users, you must create policies. See [Data Flow Policies](https://docs.oracle.com/en-us/iaas/data-flow/using/policies.htm) and [Data Catalog Metastore Required Policies](https://docs.oracle.com/en-us/iaas/data-catalog/using/metastore.htm) for more details.\n",
    "\n",
    "<a id='var'></a>\n",
    "### Variables\n",
    "\n",
    "To run this notebook, you must provide some information about your tenancy configuration. The `<job_name>` is a unique name for a job. To connect to the metastore, replace `<metastore_id>` with the OCID for the metastore. To create and run a Data Flow application, you must specify a compartment and buckets for storing logs and the Data Flow script. These resources must be in the same compartment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2b0c5749",
   "metadata": {},
   "outputs": [],
   "source": [
    "job_name = \"<job_name>\"\n",
    "log_bucket_uri = \"oci://<bucket_name>@<namespace>/<prefix>\"\n",
    "metastore_id = \"<metastore_id>\"\n",
    "script_bucket = \"oci://<bucket_name>@<namespace>/<prefix>\"\n",
    "\n",
    "compartment_id = os.environ.get(\"NB_SESSION_COMPARTMENT_OCID\")\n",
    "driver_shape = \"VM.Standard2.1\"\n",
    "executor_shape = \"VM.Standard2.1\"\n",
    "spark_version = \"3.0.2\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b184de99",
   "metadata": {},
   "source": [
    "<a id='appscript'></a>\n",
    "# Application Script\n",
    "\n",
    "An application script is used to execute the Data Flow job. The following cell creates this script and saves it to local storage. However, Data Flow requires that the script is stored in Object Storage as it cannot access your notebook session. The ADS framework takes care of uploading this script to Object Storage for you.\n",
    "\n",
    "The next cell contains the script in a single string. The script is written to the local storage. This method works well for small scripts. Larger scripts are developed outside of the notebook. The application script uses Employee Attrition data to create a new database and a product view table. This data is loaded in from a publicly accessible Object Storage bucket. The metastore manages all the metadata about the new database, while the actual data is copied to your Object Storage bucket. The script performs a query on the database. Finally, the script removes the database from the metastore. This causes the files in Object Storage, related to the database, to be removed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5c257052",
   "metadata": {},
   "outputs": [],
   "source": [
    "script = '''\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "def main():\n",
    "\n",
    "    database_name = \"employee_attrition\"\n",
    "    table_name = \"orcl_attrition\"\n",
    "\n",
    "    # Create a Spark session\n",
    "    spark = SparkSession \\\\\n",
    "        .builder \\\\\n",
    "        .appName(\"Python Spark SQL basic example\") \\\\\n",
    "        .enableHiveSupport() \\\\\n",
    "        .getOrCreate()\n",
    "\n",
    "    # Load a CSV file from a public Object Storage bucket\n",
    "    df = spark \\\\\n",
    "        .read \\\\\n",
    "        .format(\"csv\") \\\\\n",
    "        .option(\"header\", \"true\") \\\\\n",
    "        .option(\"multiLine\", \"true\") \\\\\n",
    "        .load(\"oci://hosted-ds-datasets@bigdatadatasciencelarge/synthetic/orcl_attrition.csv\")\n",
    "\n",
    "    print(f\"Creating {database_name}\")\n",
    "    spark.sql(f\"DROP DATABASE IF EXISTS {database_name} CASCADE\")\n",
    "    spark.sql(f\"CREATE DATABASE IF NOT EXISTS {database_name}\")\n",
    "\n",
    "    # Write the data to the database\n",
    "    df.write.mode(\"overwrite\").saveAsTable(f\"{database_name}.{table_name}\")\n",
    "\n",
    "    # Use Spark SQL to read from the database.\n",
    "    query_result_df = spark.sql(f\"\"\"\n",
    "                                SELECT EducationField, SalaryLevel, JobRole FROM {database_name}.{table_name} limit 10\n",
    "                                \"\"\")\n",
    "\n",
    "    # Convert the filtered Apache Spark DataFrame into JSON format and write it out to stdout\n",
    "    # so that it can be captured in the log.\n",
    "    print('\\\\n'.join(query_result_df.toJSON().collect()))\n",
    "    \n",
    "    # Clean resources\n",
    "    spark.sql(f\"DROP DATABASE IF EXISTS {database_name} CASCADE\")\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    main()\n",
    "'''\n",
    "\n",
    "dataflow_base_folder = tempfile.mkdtemp()\n",
    "print(f\"Data flow directory: {dataflow_base_folder}\")\n",
    "\n",
    "pyspark_file_path = os.path.join(dataflow_base_folder, \"example.py\")\n",
    "\n",
    "with open(pyspark_file_path, \"w\") as f:\n",
    "    print(script.strip(), file=f)\n",
    "\n",
    "print(f\"Script path: {pyspark_file_path}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e7d397d7",
   "metadata": {},
   "source": [
    "<a id='jobs'></a>\n",
    "# Create and Run a Data Flow Application\n",
    "\n",
    "<a id='conf'></a>\n",
    "## Configurating Job\n",
    "\n",
    "The preferred method for running Data Flow applications is to run them as a Job. This Job allows you to better manage your resources and isolate the Data Flow application from the notebook. A `DataFlow` object must be created and is a subclass of `Infrastructure`. The object defines the metadata related to the Data Flow service. For example, the object stores properties specific to Data Flow service, such as `compartment_id`, `logs_bucket_uri`. This object also defines the connection between Data Flow and the metastore. To define the actual parameters needed to run the Data Flow job, a `DataFlowRuntime` object is required. The object is a subclass of `Runtime`. `DataFlowRuntime` stores properties related to the script to be run. The object defines the buckets used for the logs, the location of the Data Flow application script, and any command line options needed.\n",
    "\n",
    "To use a private bucket as the `logs_bucket`, ensure that a Data Flow Service policy has been added. See the [prerequisite step](#prereq) and the [policy setup page](https://docs.cloud.oracle.com/en-us/iaas/data-flow/using/dfs_getting_started.htm#policy_set_up) for more details.\n",
    "\n",
    "In the following example, the `dataflow_configs` variable is a `DataFlow` that has the compartment OCID, metastore OCID, log bucket URI, the compute shape for the driver, the compute shape that is used for the executor, and the version of Spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "900bfbcd",
   "metadata": {},
   "outputs": [],
   "source": [
    "if metastore_id != \"<metastore_id>\":\n",
    "    dataflow_configs = DataFlow(\n",
    "        {\n",
    "            \"compartment_id\": compartment_id,\n",
    "            \"driver_shape\": driver_shape,\n",
    "            \"executor_shape\": executor_shape,\n",
    "            \"logs_bucket_uri\": log_bucket_uri,\n",
    "            \"metastore_id\": metastore_id,\n",
    "            \"spark_version\": spark_version,\n",
    "        }\n",
    "    )\n",
    "else:\n",
    "    print(\n",
    "        \"DataFlow object was not created. Enter configuration values in the Setup section.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "333e8a03",
   "metadata": {},
   "source": [
    "The `runtime_config` variable is a `DataFlowRuntime` object. It contains information about the location of the script and the bucket for the script. The script URI defines the location of the Data Flow application script. This can be on local storage or in Object Storage. If the path is local, then the script bucket must be specified so that the framework can upload the script to the Object Storage bucket. Data Flow requires a script to be available in Object Storage. The URI for buckets must have the following format `oci://<bucket_name>@<namespace>/<prefix>`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f8a06f57",
   "metadata": {},
   "outputs": [],
   "source": [
    "if metastore_id != \"<metastore_id>\":\n",
    "    runtime_config = (\n",
    "        DataFlowRuntime()\n",
    "        .with_script_uri(pyspark_file_path)\n",
    "        .with_script_bucket(script_bucket)\n",
    "    )\n",
    "else:\n",
    "    print(\n",
    "        \"DataFlow object was not created. Enter configuration values in the Setup section.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2b08d881",
   "metadata": {},
   "source": [
    "The following cell creates a Job that executes the Data Flow application. The `Job` object needs a name, information about the Data Flow cluster infrastructure, and the runtime configuration. The `.create()` method is used to create the Data Flow application."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "af6e938c",
   "metadata": {},
   "outputs": [],
   "source": [
    "if metastore_id != \"<metastore_id>\":\n",
    "    df_job = Job(name=job_name, infrastructure=dataflow_configs, runtime=runtime_config)\n",
    "    df_app = df_job.create()\n",
    "else:\n",
    "    print(\n",
    "        \"Job object was not created. Enter configuration values in the Setup section.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e2ae0c6e",
   "metadata": {},
   "source": [
    "<a id='run'></a>\n",
    "## Run the Data Flow Application\n",
    "\n",
    "To run this Data Flow application, call the `.run()` method. It creates a `DataFlowRun` object. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "edbc3945",
   "metadata": {},
   "outputs": [],
   "source": [
    "if metastore_id != \"<metastore_id>\":\n",
    "    df_run = df_app.run()\n",
    "else:\n",
    "    print(\n",
    "        \"Job object was not created. Enter configuration values in the Setup section.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9e5d5e89",
   "metadata": {},
   "source": [
    "The `.watch()` method on the `DataFlowRun` object accesses the logs and prints them to the screen."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "894db73c",
   "metadata": {},
   "outputs": [],
   "source": [
    "if metastore_id != \"<metastore_id>\":\n",
    "    df_run.watch()\n",
    "else:\n",
    "    print(\n",
    "        \"Job object was not created. Enter configuration values in the Setup section.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "34f7ff76",
   "metadata": {},
   "source": [
    "<a id='clean_up'></a>\n",
    "# Clean Up\n",
    "\n",
    "This notebook creates several resources such as a database with a metastore entry and files in Object Storage. Also, the notebook creates a Data Flow instance and a Job. The Data Flow application deletes the database, removes the entry in the Data Catalog, and deletes the files on Object Storage related to the database. The Data Flow automatically cleans up when done. You have to manually clean up the Data Flow application script and the associated log files. The following cell cleans up the Job objects."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1e089281",
   "metadata": {},
   "outputs": [],
   "source": [
    "if metastore_id != \"<metastore_id>\":\n",
    "    df_run.delete()\n",
    "else:\n",
    "    print(\"Skipping, as the metastore_id is not defined.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dee703dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "shutil.rmtree(dataflow_base_folder)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "589f489e",
   "metadata": {},
   "source": [
    "Use [ocifs](https://ocifs.readthedocs.io/en/latest/unix-operations.html#rm) to clean up the Data Flow log and script buckets."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "011ec30b",
   "metadata": {},
   "source": [
    "<a id='ref'></a>\n",
    "# References\n",
    "\n",
    "- [ADS Library Documentation](https://accelerated-data-science.readthedocs.io/en/latest/index.html)\n",
    "- [Connecting to an Autonomous Database](https://docs.oracle.com/en-us/iaas/Content/Database/Tasks/adbconnecting.htm)\n",
    "- [Data Science YouTube Videos](https://www.youtube.com/playlist?list=PLKCk3OyNwIzv6CWMhvqSB_8MLJIZdO80L)\n",
    "- [OCI Data Science Documentation](https://docs.cloud.oracle.com/en-us/iaas/data-science/using/data-science.htm)\n",
    "- [Oracle Data & AI Blog](https://blogs.oracle.com/datascience/)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.11.0 64-bit",
   "language": "python",
   "name": "python3"
  },
  "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.11.0"
  },
  "vscode": {
   "interpreter": {
    "hash": "1a1af0ee75eeea9e2e1ee996c87e7a2b11a0bebd85af04bb136d915cefc0abce"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}