notebook_examples/pyspark-data_flow-application.ipynb

{
 "cells": [
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "@notebook{pyspark-data_flow-application.ipynb,\n",
    "    title: PySpark,\n",
    "    summary: Develop local PySpark applications and work with remote clusters using Data Flow.,\n",
    "    developed_on: pyspark24_p37_cpu_v3,\n",
    "    keywords: pyspark, data flow, \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"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Oracle Data Science service sample notebook.\n",
    "\n",
    "Copyright (c) 2019, 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\">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",
    "# Overview:\n",
    "\n",
    "This notebook provides Apache Spark operations for customers by bridging the existing local PySpark workflows with cloud based capabilities. Data scientists can use their familiar local environments with JupyterLab and work with remote data and remote clusters simply by selecting a kernel. The operations that will be demonstrated are: how to use the interactive Spark environment and produce a Spark script; how to prepare and create an application; how to prepare and create a run; how to list existing dataflow applications; and how to retrieve and display the logs.\n",
    "\n",
    "The interactive Spark kernel provides a simple and efficient way to edit and build your Spark script, and easy access to read from OCI Object Storage.\n",
    "\n",
    "Compatible conda pack: [PySpark 3.2 and Data Flow](https://docs.oracle.com/iaas/data-science/using/conda-pyspark-fam.htm) for CPU on Python 3.8 (version 2.0)\n",
    "\n",
    "---\n",
    "\n",
    "## Contents:\n",
    "\n",
    "- <a href='#kernel'>Build a PySpark Script Using an Interactive Spark Kernel</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",
    "\n",
    "---\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "from pyspark.sql import SparkSession"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='kernel'></a>\n",
    "# Build a PySpark Script Using an Interactive Spark Kernel \n",
    "\n",
    "Set up Spark session in your PySpark conda environment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a Spark session\n",
    "spark = (\n",
    "    SparkSession.builder.appName(\"Python Spark SQL basic example\")\n",
    "    .config(\"spark.driver.cores\", \"4\")\n",
    "    .config(\"spark.executor.cores\", \"4\")\n",
    "    .getOrCreate()\n",
    ")\n",
    "\n",
    "spark.sparkContext.setLogLevel(\"ERROR\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load the Employee Attrition data file from Oracle Cloud Infrastructure Object Storage into an Apache Spark DataFrame. You can configure your `core-site.xml` for accessing to Object Storage by `odsc core-site config` command. Running `odsc core-site config -h` in terminal to see the usage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "emp_attrition = (\n",
    "    spark.read.format(\"csv\")\n",
    "    .option(\"header\", \"true\")\n",
    "    .option(\"inferSchema\", \"true\")\n",
    "    .option(\"multiLine\", \"true\")\n",
    "    .load(\n",
    "        \"oci://hosted-ds-datasets@bigdatadatasciencelarge/synthetic/orcl_attrition.csv\"\n",
    "    )\n",
    "    .cache()\n",
    ")  # cache the dataset to increase computing speed\n",
    "emp_attrition.createOrReplaceTempView(\"emp_attrition\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Explore the dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"select * from emp_attrition limit 5\").toPandas()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Visualize how monthly income and age relate to one another in the context of years in industry"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fig, ax = plt.subplots()\n",
    "plot = (\n",
    "    spark.sql(\n",
    "        \"\"\"\n",
    "          SELECT \n",
    "              Age,\n",
    "              MonthlyIncome,\n",
    "              YearsInIndustry\n",
    "          FROM\n",
    "            emp_attrition \n",
    "          \"\"\"\n",
    "    )\n",
    "    .toPandas()\n",
    "    .plot.scatter(\n",
    "        x=\"Age\",\n",
    "        y=\"MonthlyIncome\",\n",
    "        title=\"Age vs Monthly Income\",\n",
    "        c=\"YearsInIndustry\",\n",
    "        cmap=\"viridis\",\n",
    "        figsize=(12, 12),\n",
    "        ax=ax,\n",
    "    )\n",
    ")\n",
    "plot.set_xlabel(\"Age\")\n",
    "plot.set_ylabel(\"Monthly Income\")\n",
    "plot"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "View all of the columns in the table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"show columns from emp_attrition\").show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Select a few columns using Apache Spark and convert it into a Pandas DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = (\n",
    "    spark.sql(\n",
    "        \"\"\"\n",
    "         SELECT\n",
    "            Age,\n",
    "            MonthlyIncome,\n",
    "            YearsInIndustry\n",
    "          FROM\n",
    "            emp_attrition \"\"\"\n",
    "    )\n",
    "    .limit(10)\n",
    "    .toPandas()\n",
    ")\n",
    "df"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also work with different compression formats within Dataflow. For example snappy parquet: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Writing to a snappy parquet file\n",
    "df.to_parquet(\"emp_attrition.parquet.snappy\", compression=\"snappy\")\n",
    "pd.read_parquet(\"emp_attrition.parquet.snappy\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We are able to read in this snappy parquet file to an Apache Spark dataframe\n",
    "read_snappy_df = (\n",
    "    SparkSession.builder.appName(\"Snappy Compression Loading Example\")\n",
    "    .config(\"spark.io.compression.codec\", \"org.apache.spark.io.SnappyCompressionCodec\")\n",
    "    .getOrCreate()\n",
    "    .read.format(\"parquet\")\n",
    "    .load(f\"{os.getcwd()}/emp_attrition.parquet.snappy\")\n",
    ")\n",
    "\n",
    "read_snappy_df.first()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note: other compression formats Data Flow supports today include snappy parquet (example above) and gzip on both csv and parquet."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id='df_app'></a>\n",
    "## Create a Data Flow application\n",
    "`oracle-ads` provides different ways to submit your code to Data Flow for workloads that require larger resources. To learn more, read the [user guide](https://accelerated-data-science.readthedocs.io/en/latest/user_guide/apachespark/dataflow.html#)."
   ]
  },
  {
   "attachments": {},
   "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",
    "- [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 (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
}