notebook_examples/pyspark-data_catalog-hive_metastore.ipynb

{
 "cells": [
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "@notebook{pyspark-data_catalog-hive_metastore.ipynb,\n",
    "    title: Using Data Catalog Metastore with PySpark,\n",
    "    summary: Configure and use PySpark to process data in the Oracle Cloud Infrastructure (OCI) Data Catalog metastore, including common operations like creating and loading data from the metastore.,\n",
    "    developed_on: pyspark30_p37_cpu_v5,\n",
    "    keywords: dcat, data catalog metastore, pyspark,\n",
    "    license: Universal Permissive License v 1.0\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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",
   "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 Metastore with PySpark</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",
    "# Prerequisites\n",
    "\n",
    "Compatible conda pack: <pack> (see links for examples)\n",
    "\n",
    "***\n",
    "\n",
    "# Overview:\n",
    "\n",
    "This notebook demonstrates how to configure and use PySpark to process data in the Oracle Cloud Infrastructure (OCI) Data Catalog metastore. The [Data Catalog](https://docs.oracle.com/en-us/iaas/data-catalog/home.htm) service is a metadata management service that helps data professionals discover data and supports data governance.  The [Data Catalog Hive metastore](https://docs.oracle.com/en-us/iaas/data-catalog/using/metastore.htm) is part of the Data Catalog service and it provides schema definitions for objects in structured and unstructured data assets that reside in Object Storage. Use the metastore as a central metadata repository manage data tables that are backed by files in Object Storage. To be able to access the metastore from the notebook using PySpark, you must configure PySpark. This notebook demostrates some common operations, such as create and load tables from the metastore. It also shows you how to use PySpark to query a table and access the results in a notebook session.\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='#setup'>Setup</a>\n",
    "     - <a href='#setup_spark-defaults'>`spark-defaults.conf`</a>\n",
    "     - <a href='#setup_session'>Session Setup</a>\n",
    "     - <a href='#conda_configuration_dcat_testing'>Testing the Configuration</a>\n",
    " - <a href='#write_dcat'>Save the Data to Data Catalog Metastore</a>\n",
    " - <a href='#read_dcat'>Read Data from Data Catalog Metastore</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,
   "metadata": {},
   "outputs": [],
   "source": [
    "import fsspec\n",
    "import oci\n",
    "import os\n",
    "import pandas as pd\n",
    "import pyarrow.parquet as pq\n",
    "import shutil\n",
    "import tempfile\n",
    "\n",
    "from ads.common import auth as authutil\n",
    "from pyspark import SparkConf\n",
    "from pyspark.sql import SparkSession, Row\n",
    "from urllib.parse import urlparse"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='intro'></a>\n",
    "# Introduction\n",
    "\n",
    "Various data professionals, such as data engineers, data scientists, data stewards, and chief data officers, use [Data Catalog](https://docs.oracle.com/en-us/iaas/data-catalog/home.htm) to manage metadata. In the Data Catalog, data assets represent a data source, such as a database, an object store, a file or document store, a message queue, or an application. One of the most common uses for data scientists is to use Data Catalog to manage metadata about a flat-file database that is backed by Object Storage. This notebook demonstrates how to connect to the Data Catalog, and create and read from a database on Object Storage. This database must have its metadata stored in the Data Catalog.\n",
    "\n",
    "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 semi-structured data assets on Object Storage. The metastore is the central metadata repository to understand tables that are backed by files on object storage. The metastore provides an invocation endpoint that 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 maintained in Object Storage. [Apache Spark SQL](https://spark.apache.org/sql/) makes use of a Hive metastore for this purpose.\n",
    "\n",
    "<a id='setup'></a>\n",
    "# Setup\n",
    "\n",
    "To set up the environment, a `spark-defaults.conf` msut be configured. Several variables that define entities such as the Data Catalog Metastore and the location of the data warehouse bucket must also be defined.\n",
    "\n",
    "<a id='setup_spark-defaults'></a>\n",
    "## `spark-defaults.conf`\n",
    "\n",
    "The `spark-defaults.conf` file is used to define the properties that are used by Spark. A templated version is installed when you install a Data Science conda environment that supports PySpark. However, you must update the template so that the Data Catalog metastore can be accessed. You can do this manually. However, the `odsc data-catalog config` commandline tool is ideal for setting up the file because it gathers information about your environment, and uses that to build the file.\n",
    "\n",
    "The `odsc data-catalog config` command line tool needs the `--metastore` option to define the Data Catalog metastore OCID. No other command line option is needed because settings have default values, or they take values from your notebook session environment. Following are common parameters that you may need to override.\n",
    "\n",
    "The `--authentication` option sets the authentication mode. It supports resource principal and API keys. The preferred method for authentication is resource principal, which is sent with `--authentication resource_principal`. If you want to use API keys, then use the `--authentication api_key` option. If the `--authentication` isn't specified, API keys are used. When API keys are used, information from the OCI configuration file is used to create the `spark-defaults.conf` file.\n",
    "\n",
    "Object Storage and Data Catalog are regional services. By default, the region is set to the region your notebook session is running in. This information is taken from the environment variable, `NB_REGION`. Use the `--region` option to override this behavior.\n",
    "\n",
    "The default location of the `spark-defaults.conf` file is `/home/datascience/spark_conf_dir` as defined in the `SPARK_CONF_DIR` environment variable. Use the `--output` option to define the directory where to write the file.\n",
    "\n",
    "You need to determine what settings are appropriate for your configuration. However, the following works for most configurations and is run in a terminal window.\n",
    "\n",
    "```bash\n",
    "odsc data-catalog config --authentication resource_principal --metastore <metastore_id>\n",
    "```\n",
    "For more assistance, use the following command in a terminal window:\n",
    "\n",
    "```bash\n",
    "odsc data-catalog config --help\n",
    "```\n",
    "\n",
    "<a id='setup_session'></a>\n",
    "## Session Setup\n",
    "\n",
    "The notebook makes connections to the Data Catalog metastore and Object Storage. In the next cell, specify the bucket URI to act as the data warehouse. Use the `warehouse_uri` variable with the `oci://<bucket_name>@<namespace_name>/<key>` format. Update the variable `metastore_id` with the OCID of the Data Catalog metastore."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# warehouse_uri points to the default location for managed databases and tables\n",
    "warehouse_uri = \"<warehouse_uri>\"\n",
    "metastore_id = \"<metastore_id>\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='conda_configuration_dcat_testing'></a>\n",
    "## Testing the Configuration\n",
    "\n",
    "To test the configuration, the next cell connects to the Data Catalog metastore, and requests a list of databases:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if metastore_id != \"<metastore_id>\":\n",
    "    spark = (\n",
    "        SparkSession.builder.appName(\"Python Spark SQL Hive integration example\")\n",
    "        .config(\"spark.sql.warehouse.dir\", warehouse_uri)\n",
    "        .config(\"spark.hadoop.oracle.dcat.metastore.id\", metastore_id)\n",
    "        .enableHiveSupport()\n",
    "        .getOrCreate()\n",
    "    )\n",
    "    spark.sparkContext.setLogLevel(\"ERROR\")\n",
    "\n",
    "    # show the databases in the warehouse:\n",
    "    spark.sql(\"SHOW DATABASES\").show()\n",
    "else:\n",
    "    spark = None\n",
    "    print(\n",
    "        \"No connection was made to the Data Catalog Metastore. Enter configuration values in the Setup section.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='write_dcat'></a>\n",
    "# Save to Data Catalog Metastore\n",
    "\n",
    "In this section, the connection to the Data Catalog metastore is used to create a database in the Object Storage. The metastore manages the metadata about the table, and Object Storage manages the files. PySpark is used to perform the actual operations.\n",
    "\n",
    "In the next cell, PySpark creates a database. Since PySpark is connected to the metastore, the Data Catalog metastore manages the metadata about the table. Additionally, the PySpark connection is linked to an Object Storage bucket, so the data is written to it. From the perspective of Object Storage, it is only mananging files. The metastore doesn't perform the actual operations on the data. That is handled by PySpark, but PySpark depends on the metastore to understand that the files in Object Storage is a database.\n",
    "\n",
    "After the database is created, PySparks reads data from a publically accessible Object Storage bucket. Then it's written to the data warehouse Object Storage bucket, and the metadata in the metastore is updated."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "database_name = \"ODSC_DEMO\"\n",
    "table_name = \"ODSC_PYSPARK_METASTORE_DEMO\"\n",
    "file_path = (\n",
    "    \"oci://hosted-ds-datasets@bigdatadatasciencelarge/synthetic/orcl_attrition.csv\"\n",
    ")\n",
    "\n",
    "if spark is not None:\n",
    "    spark.sql(f\"DROP DATABASE IF EXISTS {database_name} CASCADE\")\n",
    "    spark.sql(f\"CREATE DATABASE {database_name}\")\n",
    "    input_dataframe = spark.read.option(\"header\", \"true\").csv(file_path)\n",
    "    input_dataframe.write.mode(\"overwrite\").saveAsTable(f\"{database_name}.{table_name}\")\n",
    "else:\n",
    "    print(\"Database was not created. Enter configuration values in the Setup section.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='read_dcat'></a>\n",
    "# Read from Data Catalog Metastore\n",
    "\n",
    "Once the PySpark connection has been made to the Data Catalog metastore and the Object Storage bucket that backs the data warehouse, you can perform PySpark operations similar to any other PySpark setup. The following cell uses HiveQL, which is a SQL like data manipulation language (DML) that retrieves records from a database that is managed by the Data Catalog metastore."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if spark is not None:\n",
    "    spark_df = spark.sql(\n",
    "        f\"\"\"\n",
    "                        SELECT EducationField, SalaryLevel, JobRole FROM {database_name}.{table_name} limit 10\n",
    "                        \"\"\"\n",
    "    )\n",
    "    spark_df.show()\n",
    "else:\n",
    "    spark_df = None\n",
    "    print(\n",
    "        \"No HiveQL query was executed. Enter configuration values in the Setup section.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='query_topd'></a>\n",
    "## Convert to a Pandas DataFrame\n",
    "\n",
    "The `spark_df` object doesn't actually contain the record results in memory. Generally, a PySpark dataframe is used when the data is very large. However, it is a common data science workflow pattern to perform computations in PySpark that aggrigate and reduce the dataset size so that it can be used in a notebook session. \n",
    "\n",
    "The built-in PySpark ``.toPandas()`` method is used to convert PySpark dataframe to a Pandas dataframe. It is an expensive operation that you should use carefully to minimize the performance impact on your Spark applications. If you require this, especially when the dataframe is fairly large, you need to consider `PyArrow` optimization when converting Spark to Pandas dataframe or the reverse. To use Arrow, set the Spark configuration `spark.sql.execution.arrow.enabled` to `true`. This configuration is disabled by default."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if spark is not None:\n",
    "    pd_df = spark_df.toPandas()\n",
    "    print(pd_df)\n",
    "else:\n",
    "    print(\n",
    "        \"The PySpark DataFrame was not converted to a Pandas DataFrame. Enter configuration values in the Setup section.\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='clean_up'></a>\n",
    "# Clean Up\n",
    "\n",
    "This notebook created a number of artifacts, such as creating a database table and starting a Apache Spark cluster. The next cell removes these resources."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if spark is not None:\n",
    "    spark.sql(f\"DROP DATABASE IF EXISTS {database_name} CASCADE\")\n",
    "    spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "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/)\n",
    "- [PySpark Documentation](https://spark.apache.org/docs/latest/api/python/)\n",
    "- [Using sqlnet.ora file with JDBC](https://stackoverflow.com/questions/63696611/can-the-oracle-jdbc-thin-driver-use-a-sqlnet-ora-file-for-configuration)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.15"
  },
  "vscode": {
   "interpreter": {
    "hash": "aee8b7b246df8f9039afb4144a1f6fd8d2ca17a180786b69acc140d282b71a49"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}