data-ingestion-and-preparation/spark-sql-analytics.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Spark SQL Analytics with the Iguazio Data Science Platform\n",
    "\n",
    "- [Overview](#overview)\n",
    "- [Setup](#setup)\n",
    "- [Initiate a Spark Session and Configure Spark](#initiate-a-spark-session-and-configure-spark)\n",
    "  - [Modify the Spark Configuration (Optional)](#modify-the-spark-cfg)\n",
    "- [Load Data into a Spark DataFrame](#load-data)\n",
    "  - [Load Data from Amazon S3](#load-data-from-amazon-s3)\n",
    "  - [Load Data from an External Table](#load-data-from-external-table)\n",
    "  - [Load Data from a-Semi-Structured File](#load-data-from-semi-structured-file)\n",
    "  - [Load Data from an Unstructured File](#load-data-from-unstructured-file)\n",
    "  - [Overwrite the Table Schema](#overwrite-table-schema)\n",
    "- [Use Spark SQL](#spark-sql)\n",
    "  - [Spark SQL on an Object](#spark-sql-on-object)\n",
    "  - [Spark SQL on a Table](#spark-sql-on-table)\n",
    "  - [Spark SQL on Platform NoSQL Data](#spark-sql-on-platform-nosql-data)\n",
    "  - [Spark SQL Join](#spark-sql-join)\n",
    "  - [Spark SQL on a Parquet File](#spark-sql-on-parquet)\n",
    "  - [Spark SQL on a Partitioned Table](#spark-sql-on-partitioned-table)\n",
    "- [Perform Conditional Data Updates](#conditional-update)\n",
    "- [Cleanup](#Cleanup)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"overview\"></a>\n",
    "## Overview\n",
    "\n",
    "Spark SQL is an Apache Spark module for working with structured data.\n",
    "IT lets you query structured data inside Spark programs by using either SQL or a familiar DataFrame API.\n",
    "DataFrames and SQL provide a common way to access a variety of data sources.\n",
    "\n",
    "In this notebook, you'll learn how to use Spark SQL and DataFrames to access objects, tables, and unstructured data that persist in the data containers of the [Iguazio Data Science Platform](https://www.iguazio.com/) (**the platform**).\n",
    "\n",
    "The platform's Spark drivers implement the data-source API and support predicate push down: the queries are passed to the platform's data store, which returns only the relevant data.\n",
    "This allow accelerated and high-speed access from Spark to data stored in the platform.\n",
    "\n",
    "For more, details read the [Spark SQL and DataFrames documentation](https://spark.apache.org/docs/2.3.1/sql-programming-guide.html) and the overview in platform's [Spark APIs Reference](https://www.iguazio.com/docs/v3.0/data-layer/reference/spark-apis/overview/)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"setup\"></a>\n",
    "## Setup\n",
    "\n",
    "Before preparing the data, you need to define some environment variables that will be used in the following steps of this tutorial:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "env: DIR1=examples\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "\n",
    "## Iguazio Data Science Platform Variables\n",
    "# Directory for stocks\n",
    "%env DIR1 = examples"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"load-data-from-amazon-s3\"></a>\n",
    "### Load Data from Amazon S3\n",
    "\n",
    "Load a file from an Amazon S3 bucket into a Spark DataFrame.<br>\n",
    "The URL of the S3 file should be of the form `s3a://bucket/path/to/file`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2020-04-07 22:41:46--  https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk/1.7.4/aws-java-sdk-1.7.4.jar\n",
      "Resolving repo1.maven.org (repo1.maven.org)... 151.101.200.209\n",
      "Connecting to repo1.maven.org (repo1.maven.org)|151.101.200.209|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 11948376 (11M) [application/java-archive]\n",
      "Saving to: ‘/spark/jars/aws-java-sdk-1.7.4.jar.7’\n",
      "\n",
      "aws-java-sdk-1.7.4. 100%[===================>]  11.39M  25.6MB/s    in 0.4s    \n",
      "\n",
      "2020-04-07 22:41:47 (25.6 MB/s) - ‘/spark/jars/aws-java-sdk-1.7.4.jar.7’ saved [11948376/11948376]\n",
      "\n",
      "--2020-04-07 22:41:47--  https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/2.7.3/hadoop-aws-2.7.3.jar\n",
      "Resolving repo1.maven.org (repo1.maven.org)... 151.101.200.209\n",
      "Connecting to repo1.maven.org (repo1.maven.org)|151.101.200.209|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 126287 (123K) [application/java-archive]\n",
      "Saving to: ‘/spark/jars/hadoop-aws-2.7.3.jar.7’\n",
      "\n",
      "hadoop-aws-2.7.3.ja 100%[===================>] 123.33K  --.-KB/s    in 0.03s   \n",
      "\n",
      "2020-04-07 22:41:47 (3.55 MB/s) - ‘/spark/jars/hadoop-aws-2.7.3.jar.7’ saved [126287/126287]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!wget https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk/1.7.4/aws-java-sdk-1.7.4.jar -P $SPARK_HOME/jars/\n",
    "!wget https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/2.7.3/hadoop-aws-2.7.3.jar -P $SPARK_HOME/jars/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "sdf\n",
      "DataFrame[referenceId: string, custaccountcustid: string, custaccountacctid: int, modcust: string, moduser: string, moddatetime: bigint, recordowner: int, diis_acctnum: bigint, diis_accthidden: string, diis_mapdate: string, cif_id: int, diis_ausr: string]\n"
     ]
    }
   ],
   "source": [
    "import botocore.session\n",
    "from pyspark import SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "session = botocore.session.get_session()\n",
    "credentials = session.get_credentials()\n",
    "\n",
    "print(\"sdf\")\n",
    "AWS_ACCESS_KEY=\"\"\n",
    "AWS_SECRET_KEY=\"\"\n",
    "\n",
    "conf = (\n",
    "    SparkConf()\n",
    "    .set(\"spark.hadoop.fs.s3a.path.style.access\", True)\n",
    "    .set(\"spark.hadoop.fs.s3a.access.key\", AWS_ACCESS_KEY)\n",
    "    .set(\"spark.hadoop.fs.s3a.secret.key\", AWS_SECRET_KEY)\n",
    "    .set(\"spark.hadoop.fs.s3a.endpoint\", f\"s3-us-east-2.amazonaws.com\")\n",
    "    .set(\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")\n",
    "    .set(\"com.amazonaws.services.s3.enableV4\", True)\n",
    "    .set(\"spark.driver.extraJavaOptions\", \"-Dcom.amazonaws.services.s3.enableV4=true\")\n",
    ")\n",
    "spark = SparkSession.builder.config(conf=conf).getOrCreate()\n",
    "\n",
    "s3Obj=\"s3a://bucket-s3/your-scv*\"\n",
    "df = spark.read \\\n",
    "    .format(\"csv\") \\\n",
    "    .option(\"header\", \"true\") \\\n",
    "    .option(\"inferSchema\", \"true\") \\\n",
    "    .load(s3Obj)\n",
    "print(df)\n",
    "\n",
    "# Stop Spark Session\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"initiate-a-spark-session-and-configure-spark\"></a>\n",
    "## Initiate a Spark Session and Configure Spark\n",
    "\n",
    "Begin by initiating a new Spark session and checking the default Spark configuration for this session:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "## Initial Spark Configuration ##\n",
      "\n",
      "spark.app.name =  Spark Session with Default Configurations\n",
      "spark.driver.cores =  None\n",
      "spark.driver.memory =  None\n",
      "spark.executor.cores =  None\n",
      "spark.executor.memory =  None\n",
      "spark.cores.max =  None\n",
      "spark.python.profile =  None\n",
      "spark.pyspark.python =  None\n",
      "spark.hadoop.fs.s3a.impl =  org.apache.hadoop.fs.s3a.S3AFileSystem\n",
      "CPU times: user 17.4 ms, sys: 7.15 ms, total: 24.6 ms\n",
      "Wall time: 126 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "from pyspark.conf import SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import concat, col\n",
    "\n",
    "# Initiate a new Spark Session\n",
    "spark = SparkSession.builder.appName(\"Spark Session with Default Configurations\").getOrCreate()\n",
    "\n",
    "# Retrieve and view all the default Spark configurations:\n",
    "# conf = spark.sparkContext._conf.getAll()\n",
    "# print(conf)\n",
    "conf = spark.sparkContext._conf\n",
    "\n",
    "print('\\n\\n## Initial Spark Configuration ##\\n')\n",
    "\n",
    "print('spark.app.name = ', conf.get(\"spark.app.name\"))\n",
    "print('spark.driver.cores = ', conf.get(\"spark.driver.cores\"))\n",
    "print('spark.driver.memory = ', conf.get(\"spark.driver.memory\"))\n",
    "print('spark.executor.cores = ', conf.get(\"spark.executor.cores\"))\n",
    "print('spark.executor.memory = ', conf.get(\"spark.executor.memory\"))\n",
    "print('spark.cores.max = ', conf.get(\"spark.cores.max\"))\n",
    "\n",
    "print('spark.python.profile = ', conf.get(\"spark.python.profile\"))\n",
    "print('spark.pyspark.python = ', conf.get(\"spark.pyspark.python\"))\n",
    "print('spark.hadoop.fs.s3a.impl = ', conf.get(\"spark.hadoop.fs.s3a.impl\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true
   },
   "source": [
    "<a id=\"modify-the-spark-cfg\"></a>\n",
    "### Modify the Spark Configuration (Optional)\n",
    "\n",
    "You may need to modify the default Spark configuration to match your specific requirements and resources and optimize performance.\n",
    "The nature of your datasets and data models, the data-access methods that you select to use, and your hardware resources are all relevant factors in selecting your configuration.\n",
    "The [Test the SQL Performance on a Partitioned NoSQL Table with Different Spark Configurations](#test-sql-perf-on-partitioned-nosql-table-w-different-spark-cfgs) section of this tutorial demonstrates how to test Spark SQL performance on a partitioned NoSQL table in the platform with different Spark configurations.\n",
    "\n",
    "The following Spark configuration properties are especially worth noting:\n",
    "- `spark.driver.cores`\n",
    "- `spark.driver.memory`\n",
    "- `spark.executor.cores`\n",
    "- `spark.executor.memory`\n",
    "- `spark.cores.max`\n",
    "- `spark.python.profile`\n",
    "- `spark.pyspark.python`\n",
    "\n",
    "To access data in an AWS S3 bucket, perform the following configurations; replace the `<...>` placeholders with your specific data:\n",
    "- `spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem`\n",
    "- `spark.hadoop.fs.s3a.access.key=<your access key>`\n",
    "- `spark.hadoop.fs.s3a.secret.key=<your secret key>`\n",
    "- `spark.hadoop.fs.s3a.fast.upload=true`\n",
    "\n",
    "For detailed information about configuring Spark, read the [Spark documentation](https://spark.apache.org/docs/2.3.1/configuration.html#dynamically-loading-spark-properties).<br>\n",
    "For further performance services and support, contact Iguazio's [customer-success team](https://www.iguazio.com/support/).\n",
    "\n",
    "The following code demonstrates how to modify the Spark configuration."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "\n",
      "## Modified Spark Configuration ##\n",
      "\n",
      "spark.app.name =  Spark SQL for Analytics\n",
      "spark.driver.cores =  None\n",
      "spark.driver.memory =  2g\n",
      "spark.executor.cores =  2\n",
      "spark.executor.memory =  4g\n",
      "spark.cores.max =  3\n",
      "spark.python.profile =  true\n",
      "spark.pyspark.python =  true\n",
      "spark.hadoop.fs.s3a.impl =  org.apache.hadoop.fs.s3a.S3AFileSystem\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Modify the default Spark configurations, as needed.\n",
    "# The following example uses a single m5.2xlarge application node (8 CPUs, 32 GB):\n",
    "'''\n",
    "conf = spark.sparkContext._conf\\\n",
    "    .set(\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")\n",
    "'''\n",
    "conf = spark.sparkContext._conf\\\n",
    "    .setAll([('spark.app.name', 'Spark SQL for Analytics'), \\\n",
    "             # ('spark.driver.cores', '2'), \\ # Only in cluster mode.\n",
    "             ('spark.driver.memory','2g'), \n",
    "             ('spark.executor.cores', '2'), \\\n",
    "             ('spark.executor.memory', '4g'), \\\n",
    "             ('spark.cores.max', '3'), \\\n",
    "             ('spark.python.profile', 'true'), \\\n",
    "             ('spark.pyspark.python', 'true'), \\\n",
    "             (\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")])\n",
    "\n",
    "# Stop the current Spark Session\n",
    "spark.sparkContext.stop()\n",
    "\n",
    "# Create a Spark Session with new configurations\n",
    "spark = SparkSession.builder.config(conf=conf).appName(\"Spark SQL Analytics - ipynb\").getOrCreate()\n",
    "\n",
    "print('\\n\\n## Modified Spark Configuration ##\\n')\n",
    "\n",
    "print('spark.app.name = ', conf.get(\"spark.app.name\"))\n",
    "print('spark.driver.cores = ', conf.get(\"spark.driver.cores\"))\n",
    "print('spark.driver.memory = ', conf.get(\"spark.driver.memory\"))\n",
    "print('spark.executor.cores = ', conf.get(\"spark.executor.cores\"))\n",
    "print('spark.executor.memory = ', conf.get(\"spark.executor.memory\"))\n",
    "print('spark.cores.max = ', conf.get(\"spark.cores.max\"))\n",
    "\n",
    "print('spark.python.profile = ', conf.get(\"spark.python.profile\"))\n",
    "print('spark.pyspark.python = ', conf.get(\"spark.pyspark.python\"))\n",
    "print('spark.hadoop.fs.s3a.impl = ', conf.get(\"spark.hadoop.fs.s3a.impl\"))\n",
    "\n",
    "print('\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": false
   },
   "source": [
    "<a id=\"load-data\"></a>\n",
    "## Load Data into a Spark DataFrame\n",
    "\n",
    "The Spark Data Sources API supports a pluggable mechanism for integration with structured data sources.  It is a unified API designed to support two major operations:\n",
    "\n",
    "1. Loading structured data from an external data source into Spark.\n",
    "2. Storing structured data from Spark into an external data source."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true,
    "toc-nb-collapsed": true
   },
   "source": [
    "#### Copy a File from an AWS S3 Bucket to the Platform\n",
    "\n",
    "Alternatively, you can first copy the data to a platform data container."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Create a Directory in a Platform Data Container\n",
    "\n",
    "Create a directory (`DIR1`) in your user home directory in the \"users\" platform data container (`V3IO_HOME`)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "!mkdir -p /v3io/${V3IO_HOME}/${DIR1}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Copy a CSV file from an S3 Bucket to the Platform\n",
    "\n",
    "Copy a CSV file from a Simple Storage (S3) bucket to a **stocks.csv** file in a platform data container."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current\n",
      "                                 Dload  Upload   Total   Spent    Left  Speed\n",
      "100  861k  100  861k    0     0  4285k      0 --:--:-- --:--:-- --:--:-- 4264k\n"
     ]
    }
   ],
   "source": [
    "!curl -L \"https://s3.wasabisys.com/iguazio/data/stocks/2018-03-26_BINS_XETR08.csv\" > /v3io/${V3IO_HOME}/${DIR1}/stocks.csv"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### List Files in a Platform Data-Container Directory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 0\n",
      "-rw-r--r-- 1 50 nogroup 113629 Mar 25 21:05 userdata1.parquet\n",
      "drwxr-xr-x 2 50 nogroup      0 Apr  7 12:46 multiple-parquet-files\n",
      "drwxr-xr-x 2 50 nogroup      0 Apr  7 12:46 spark-output\n",
      "drwxr-xr-x 2 50 nogroup      0 Apr  7 14:30 csvs\n",
      "drwxr-xr-x 2 50 nogroup      0 Apr  7 14:32 csvc\n",
      "-rw-r--r-- 1 50 nogroup 882055 Apr  7 15:05 demo.csv\n",
      "drwxrwxrwx 2 50 nogroup      0 Apr  7 22:36 stocks_kv\n",
      "drwxrwxrwx 2 50 nogroup      0 Apr  7 22:36 stocks_kv_partition\n",
      "drwxrwxrwx 2 50 nogroup      0 Apr  7 22:39 mytable\n",
      "-rw-r--r-- 1 50 nogroup 882055 Apr  7 22:41 stocks.csv\n"
     ]
    }
   ],
   "source": [
    "!ls -altr /v3io/${V3IO_HOME}/${DIR1}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define Platform File-Path Variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "file_path = os.path.join(os.getenv('V3IO_HOME_URL')+'/examples')\n",
    "file = os.path.join(file_path+'/stocks.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load a File from a Platform Data Container into a Spark DataFrame\n",
    "\n",
    "Read the CSV file that you saved to the platform data container into a Spark DataFrame.<br>\n",
    "The following code example uses the `inferSchema` option to automatically infer the schema of the read data (recommended).\n",
    "Alternatively, you can define the schema manually:\n",
    "\n",
    "```python\n",
    "schema = StructType([\n",
    "    StructField(\"<field name>\", <field type>, <is Null>),\n",
    "   ...])\n",
    "df = spark.read.schema(schema)\n",
    "...\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.23 ms, sys: 206 µs, total: 2.43 ms\n",
      "Wall time: 1.35 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "df = spark.read\\\n",
    "    .format(\"csv\")\\\n",
    "    .option(\"header\", \"true\")\\\n",
    "    .option(\"inferSchema\", \"true\")\\\n",
    "    .load(file)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Print the Schema"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ISIN: string (nullable = true)\n",
      " |-- Mnemonic: string (nullable = true)\n",
      " |-- SecurityDesc: string (nullable = true)\n",
      " |-- SecurityType: string (nullable = true)\n",
      " |-- Currency: string (nullable = true)\n",
      " |-- SecurityID: integer (nullable = true)\n",
      " |-- Date: timestamp (nullable = true)\n",
      " |-- Time: string (nullable = true)\n",
      " |-- StartPrice: double (nullable = true)\n",
      " |-- MaxPrice: double (nullable = true)\n",
      " |-- MinPrice: double (nullable = true)\n",
      " |-- EndPrice: double (nullable = true)\n",
      " |-- TradedVolume: integer (nullable = true)\n",
      " |-- NumberOfTrades: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### List Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['ISIN',\n",
       " 'Mnemonic',\n",
       " 'SecurityDesc',\n",
       " 'SecurityType',\n",
       " 'Currency',\n",
       " 'SecurityID',\n",
       " 'Date',\n",
       " 'Time',\n",
       " 'StartPrice',\n",
       " 'MaxPrice',\n",
       " 'MinPrice',\n",
       " 'EndPrice',\n",
       " 'TradedVolume',\n",
       " 'NumberOfTrades']"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": false
   },
   "source": [
    "<a id=\"load-data-from-external-table\"></a>\n",
    "### Load Data from an External Table\n",
    "\n",
    "In this section,  let's walk through two examples:\n",
    "\n",
    "1. Use the PyMySQL Python MySQL client library and a pandas DataFrame to load data from a MySQL database.\n",
    "2. Use Spark JDBC to read a table from AWS Redshift.\n",
    "\n",
    "\n",
    "For more details read [read-external-db](read-external-db.ipynb) and [Spark JDBC to Databases](SparkJDBCtoDBs.ipynb)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true
   },
   "source": [
    "<a id=\"load-data-from-external-table-mysql\"></a>\n",
    "#### Use MySQL as an External Data Source"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Create a MySQL Database Connection\n",
    "\n",
    "Read from a MySQL database as a bulk operation using pandas DataFrames.\n",
    "\n",
    "> **AWS Cloud Note:** If you're running the notebook code from the AWS cloud, note that AWS S3 provides **eventual consistency**.\n",
    "Therefore, it takes time for users using the persisted data and software package."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>rfam_acc</th>\n",
       "      <th>rfam_id</th>\n",
       "      <th>auto_wiki</th>\n",
       "      <th>description</th>\n",
       "      <th>author</th>\n",
       "      <th>seed_source</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>3013</th>\n",
       "      <td>RF03113</td>\n",
       "      <td>Poribacteria-1</td>\n",
       "      <td>2702</td>\n",
       "      <td>Poribacteria-1 RNA</td>\n",
       "      <td>Weinberg Z</td>\n",
       "      <td>Weinberg Z</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3014</th>\n",
       "      <td>RF03114</td>\n",
       "      <td>RT-1</td>\n",
       "      <td>2572</td>\n",
       "      <td>RT-1 RNA</td>\n",
       "      <td>Weinberg Z</td>\n",
       "      <td>Weinberg Z</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3015</th>\n",
       "      <td>RF03115</td>\n",
       "      <td>KDPG-aldolase</td>\n",
       "      <td>2703</td>\n",
       "      <td>KDPG-aldolase RNA</td>\n",
       "      <td>Weinberg Z</td>\n",
       "      <td>Weinberg Z</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "     rfam_acc         rfam_id  auto_wiki         description      author  \\\n",
       "3013  RF03113  Poribacteria-1       2702  Poribacteria-1 RNA  Weinberg Z   \n",
       "3014  RF03114            RT-1       2572            RT-1 RNA  Weinberg Z   \n",
       "3015  RF03115   KDPG-aldolase       2703   KDPG-aldolase RNA  Weinberg Z   \n",
       "\n",
       "     seed_source  \n",
       "3013  Weinberg Z  \n",
       "3014  Weinberg Z  \n",
       "3015  Weinberg Z  "
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import os\n",
    "import pymysql\n",
    "import pandas as pd \n",
    "\n",
    "conn = pymysql.connect(\n",
    "    host=os.getenv('DB_HOST','mysql-rfam-public.ebi.ac.uk'),\n",
    "    port=int(4497),\n",
    "    user=os.getenv('DB_USER','rfamro'),\n",
    "    passwd=os.getenv('DB_PASSWORD',''),\n",
    "    db=os.getenv('DB_NAME','Rfam'),\n",
    "    charset='utf8mb4')\n",
    "\n",
    "pdfMySQL = pd.read_sql_query(\"select rfam_acc,rfam_id,auto_wiki,description,author,seed_source FROM family\",\n",
    "    conn) \n",
    "\n",
    "pdfMySQL.tail(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Create a Spark DataFrame from a pandas DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfMySQL = spark.createDataFrame(pdfMySQL)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Display Table Records\n",
    "\n",
    "Display a few records of the \"family\" table that was read into the `dfMySQL` DataFrame in the previous steps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+---------+-------------------+--------------------+--------------------+\n",
      "|rfam_acc|  rfam_id|auto_wiki|        description|              author|         seed_source|\n",
      "+--------+---------+---------+-------------------+--------------------+--------------------+\n",
      "| RF00001|  5S_rRNA|     1302|   5S ribosomal RNA|Griffiths-Jones S...|Szymanski et al, ...|\n",
      "| RF00002|5_8S_rRNA|     1303| 5.8S ribosomal RNA|Griffiths-Jones S...|Wuyts et al, Euro...|\n",
      "| RF00003|       U1|     1304|U1 spliceosomal RNA|Griffiths-Jones S...|Zwieb C, The uRNA...|\n",
      "| RF00004|       U2|     1305|U2 spliceosomal RNA|Griffiths-Jones S...|The uRNA database...|\n",
      "| RF00005|     tRNA|     1306|               tRNA|Eddy SR, Griffith...|             Eddy SR|\n",
      "+--------+---------+---------+-------------------+--------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfMySQL.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Print the Table Schema\n",
    "\n",
    "Print the schema of the \"family\" table that was read into the `dfMySQL` DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- rfam_acc: string (nullable = true)\n",
      " |-- rfam_id: string (nullable = true)\n",
      " |-- auto_wiki: long (nullable = true)\n",
      " |-- description: string (nullable = true)\n",
      " |-- author: string (nullable = true)\n",
      " |-- seed_source: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfMySQL.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Register as a Table for Spark SQL Queries\n",
    "\n",
    "Define a temporary Spark view for running Spark SQL queries on the \"family\" table that was read into the `dfMySQL` DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfMySQL.createOrReplaceTempView(\"family\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Count Table Records\n",
    "\n",
    "Use Spark SQL to count the number records in the \"family\" table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|    3016|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT COUNT(*) FROM family\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Check for a Unique Key\n",
    "\n",
    "Check whether the `auto_wiki` column can serve as a unique key (attribute) of the \"family\" table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------+\n",
      "|count(DISTINCT auto_wiki)|\n",
      "+-------------------------+\n",
      "|                     1345|\n",
      "+-------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT COUNT(distinct(auto_wiki)) FROM family\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true
   },
   "source": [
    "<a id=\"load-data-from-external-table-amazon-redshift\"></a>\n",
    "#### Use Amazon Redshift as an External Data Source\n",
    "\n",
    "The **spark-redshift** library is a data source API for [Amazon Redshift](https://aws.amazon.com/redshift/).\n",
    "\n",
    "**Spark driver to Redshift:** The Spark driver connects to Redshift via JDBC using a username and password.\n",
    "Redshift doesn't support the use of IAM roles to authenticate this connection.\n",
    "\n",
    "**Spark to AWS S3:** S3 acts as a middleman to store bulk data when reading from or writing to Redshift."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#####  Create an Amazon S3 Bucket\n",
    "\n",
    "Create an Amazon S3 bucket named \"redshift-spark\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "tmpS3Dir = \"s3n://redshift-spark/tmp/\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Set Up Your Redshift Environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "redshiftDBName = '<your_redshift_DB_name>'\n",
    "redshiftTableName = '<your_redshift_Table_name>'\n",
    "redshiftUserId = '<your_redshift_User_ID>'\n",
    "redshiftPassword = '<your_redshift_Password>'\n",
    "redshifturl = '<your_redshift_URL>'\n",
    "jdbcURL = f\"jdbc:redshift://$redshifturl/$redshiftDBName?user=$redshiftUserId&password=$redshiftPassword\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Load a Redshift Table into a Spark DataFrame\n",
    "\n",
    "The `.format(\"com.databricks.spark.redshift\")` line tells the Spark Data Sources API that you're using the **spark-redshift** package.<br>\n",
    "Enable **spark-redshift** to use the **tmpS3Dir** temporary location in the S3 bucket to store temporary files generated by **spark-redshift**."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfRDSHFT = spark.read \\\n",
    "    .format(\"com.databricks.spark.redshift\")\\\n",
    "    .option(\"url\",jdbcURL )\\\n",
    "    .option(\"tempdir\", tmpS3Dir)\\\n",
    "    .option(\"dbtable\", redshiftTableName)\\\n",
    "    .load()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Check the Table\n",
    "\n",
    "Print the table schema and show a few records.<br>\n",
    "`spark-redshift` automatically reads the schema from the Redshift table and maps its types back to Spark SQL's types."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfRDSHFT.printSchema()\n",
    "dfRDSHFT.show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Persist the Redshift Table Data into the Platform's NoSQL Store"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfRDSHFT = spark.write\\\n",
    "    .format(\"io.iguaz.v3io.spark.sql.kv\")\\\n",
    "    .mode(\"append\")\\\n",
    "    .option(\"key\", key)\\\n",
    "    .option(\"sorting-key\", sorting-key)\\\n",
    "    .option(\"allow-overwrite-schema\", \"true\")\\\n",
    "    .save(os.path.join(os.getenv('V3IO_HOME'))+'/rdshfttable/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"load-data-from-semi-structured-file\"></a>\n",
    "### Load Data from a-Semi-Structured File"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Replace PATH_TO_A_JSON by the full URL of a JSON file, and remove the comment sign.\n",
    "# dfJSON = spark.read.json(\"PATH_TO_A_JSON\")\n",
    "\n",
    "jsonFile = os.path.join(os.getenv('V3IO_HOME_URL')+'/examples/mLines.json')\n",
    "\n",
    "dfJSON = spark.read \\\n",
    "    .option(\"multiline\", \"true\") \\\n",
    "    .json(jsonFile )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfJSON.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"load-data-from-unstructured-file\"></a>\n",
    "### Load Data from an Unstructured File\n",
    "\n",
    "> **Note:** Beginning with version 2.4, Spark supports loading images."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Replace PATH_TO_AN_IMAGE by the full URL of a JSON file, and remove the comment sign.\n",
    "# dfImage = spark.read.format(\"image\").option(\"dropInvalid\", true).load(\"PATH_TO_AN_IMAGE\")\n",
    "\n",
    "imageFile = os.path.join(os.getenv('V3IO_HOME_URL')+'/examples/CoffeeTime.jpg')\n",
    "\n",
    "dfImage = spark.read.format(\"image\").option(\"dropInvalid\", \"true\").load(imageFile)\n",
    "\n",
    "dfImage.select(\"image.origin\", \"image.width\", \"image.height\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"overwrite-table-schema\"></a>\n",
    "### Overwrite the Table Schema\n",
    "\n",
    "The following example creates a table named mytable with AttrA and AttrB attributes of type string and an AttrC attribute of type long, and then overwrites the table schema to change the type of AttrC to double:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfOWSchema = spark.createDataFrame([\n",
    "    (\"a\", \"z\", 123),\n",
    "    (\"b\", \"y\", 456)\n",
    "], [\"AttrA\", \"AttrB\", \"AttrC\"])\n",
    "dfOWSchema.write.format(\"io.iguaz.v3io.spark.sql.kv\") \\\n",
    "    .mode(\"overwrite\") \\\n",
    "    .option(\"key\", \"AttrA\") \\\n",
    "    .save(os.path.join(file_path)+'/mytable/')\n",
    "    \n",
    "dfOWSchema = spark.createDataFrame([\n",
    "    (\"c\", \"x\", 32.12),\n",
    "    (\"d\", \"v\", 45.2)\n",
    "], [\"AttrA\", \"AttrB\", \"AttrC\"])\n",
    "dfOWSchema.write.format(\"io.iguaz.v3io.spark.sql.kv\") \\\n",
    "    .mode(\"append\") \\\n",
    "    .option(\"key\", \"AttrA\") \\\n",
    "    .option(\"allow-overwrite-schema\", \"true\") \\\n",
    "    .save(os.path.join(file_path)+'/mytable/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true
   },
   "source": [
    "<a id=\"spark-sql\"></a>\n",
    "## Use Spark SQL\n",
    "\n",
    "Now, some Spark SQL queries to analyze the dataset that was loaded into `df` Spark DataFrame.<br>\n",
    "The first SQL queries list a few lines of selected columns in the dataset and retrieve some statistics of numerical columns."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": false
   },
   "source": [
    "<a id=\"spark-sql-on-object\"></a>\n",
    "### Spark SQL on an Object"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------+--------------------+------------+\n",
      "|        ISIN|Mnemonic|        SecurityDesc|SecurityType|\n",
      "+------------+--------+--------------------+------------+\n",
      "|CH0038389992|    BBZA|BB BIOTECH NAM.  ...|Common stock|\n",
      "|CH0038863350|    NESR|NESTLE NAM.      ...|Common stock|\n",
      "|LU0378438732|    C001|COMSTAGE-DAX UCIT...|         ETF|\n",
      "+------------+--------+--------------------+------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select(\"ISIN\", \"Mnemonic\", \"SecurityDesc\", \"SecurityType\").show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Retrieve Data from the First Rows"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(ISIN='CH0038389992', Mnemonic='BBZA', SecurityDesc='BB BIOTECH NAM.   SF 0,20'),\n",
       " Row(ISIN='CH0038863350', Mnemonic='NESR', SecurityDesc='NESTLE NAM.        SF-,10'),\n",
       " Row(ISIN='LU0378438732', Mnemonic='C001', SecurityDesc='COMSTAGE-DAX UCITS ETF I')]"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select(\"ISIN\", \"Mnemonic\", \"SecurityDesc\").head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Summary and Descriptive Statistics\n",
    "\n",
    "The function `describe` returns a DataFrame containing information such as the number of non-null entries (`count`), mean, standard deviation (`stddev`), and the minimum (`min`) and maximum (`max`) values for each numerical column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|      TradedVolume|\n",
      "+-------+------------------+\n",
      "|  count|              7401|\n",
      "|   mean|3035.7574652074045|\n",
      "| stddev|18191.489026530675|\n",
      "|    min|                 0|\n",
      "|    max|            839200|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.describe(\"TradedVolume\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true
   },
   "source": [
    "<a id=\"spark-sql-on-table\"></a>\n",
    "### Spark SQL on a Table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Register a Table View for Further Analytics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.createOrReplaceTempView(\"stock\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Select a Few Columns and Only Print a Few Lines"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+--------------------+----------+\n",
      "|        ISIN|        SecurityDesc|SecurityID|\n",
      "+------------+--------------------+----------+\n",
      "|CH0038389992|BB BIOTECH NAM.  ...|   2504244|\n",
      "|CH0038863350|NESTLE NAM.      ...|   2504245|\n",
      "|LU0378438732|COMSTAGE-DAX UCIT...|   2504271|\n",
      "+------------+--------------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "q = spark.sql(\"SELECT ISIN, SecurityDesc, SecurityID FROM stock limit 3\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Analyze Data to Identify Unique-Key Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|count(ISIN)|\n",
      "+-----------+\n",
      "|       7401|\n",
      "+-----------+\n",
      "\n",
      "+--------------------+\n",
      "|count(DISTINCT ISIN)|\n",
      "+--------------------+\n",
      "|                 737|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "q1 = spark.sql(\"SELECT COUNT(ISIN) FROM stock\").show()\n",
    "q2 = spark.sql(\"SELECT COUNT(DISTINCT(ISIN)) FROM stock\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|count(SecurityID)|\n",
      "+-----------------+\n",
      "|             7401|\n",
      "+-----------------+\n",
      "\n",
      "+--------------------------+\n",
      "|count(DISTINCT SecurityID)|\n",
      "+--------------------------+\n",
      "|                       737|\n",
      "+--------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "q4 = spark.sql(\"SELECT COUNT(SecurityID) FROM stock\").show()\n",
    "q5 = spark.sql(\"SELECT COUNT(DISTINCT(SecurityID)) FROM stock\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A combination of `ISIN`, `Date`, and `Time` can serve as a unqiue key:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------------------------------------+\n",
      "|count(DISTINCT named_struct(ISIN, ISIN, Date, Date, Time, Time))|\n",
      "+----------------------------------------------------------------+\n",
      "|                                                            7401|\n",
      "+----------------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "q6 = spark.sql(\"SELECT COUNT(DISTINCT(ISIN, Date, Time)) FROM stock\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Concatenate Date and Time Columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(concat(Date, Time)='2018-03-26 00:00:0008:00'),\n",
       " Row(concat(Date, Time)='2018-03-26 00:00:0008:00')]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select(concat(col(\"Date\"), col(\"Time\"))).head(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Date=datetime.datetime(2018, 3, 26, 0, 0), Time='08:00', datetime='2018-03-26 00:00:0008:00'),\n",
       " Row(Date=datetime.datetime(2018, 3, 26, 0, 0), Time='08:00', datetime='2018-03-26 00:00:0008:00'),\n",
       " Row(Date=datetime.datetime(2018, 3, 26, 0, 0), Time='08:00', datetime='2018-03-26 00:00:0008:00')]"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.withColumn(\"datetime\", concat(col(\"Date\"), col(\"Time\")))\\\n",
    "    .select(\"Date\", \"Time\", \"datetime\").head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Register Another Table with a Unique Key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.withColumn(\"datetime\", concat(df[\"Date\"], df[\"Time\"])).createOrReplaceTempView(\"stock_UUID\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Verify that the Key is Unique"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------------------------------+\n",
      "|count(DISTINCT named_struct(ISIN, ISIN, datetime, datetime))|\n",
      "+------------------------------------------------------------+\n",
      "|                                                        7401|\n",
      "+------------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "q7 = spark.sql(\"SELECT COUNT(DISTINCT(ISIN, datetime)) FROM stock_UUID\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Get Distinct Values on a Column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------+\n",
      "|count(DISTINCT datetime)|\n",
      "+------------------------+\n",
      "|                      60|\n",
      "+------------------------+\n",
      "\n",
      "CPU times: user 2.17 ms, sys: 171 µs, total: 2.34 ms\n",
      "Wall time: 327 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "q8 = spark.sql(\"SELECT COUNT(DISTINCT(datetime)) FROM stock_UUID\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Results show that **all data in this dataset is of the same date.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2 µs, sys: 0 ns, total: 2 µs\n",
      "Wall time: 5.25 µs\n",
      "+--------------------+\n",
      "|count(DISTINCT Time)|\n",
      "+--------------------+\n",
      "|                  60|\n",
      "+--------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "%time\n",
    "q9 = spark.sql(\"SELECT COUNT(DISTINCT(Time)) FROM stock_UUID\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": false
   },
   "source": [
    "<a id=\"spark-sql-on-platform-nosql-data\"></a>\n",
    "### Spark SQL on Platform NoSQL Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Persist Data from a Spark DataFrame to a Platform NoSQL Table\n",
    "\n",
    "The following code demonstrates how to write data from a Spark DataFrame to a NoSQL table in the persistent memory in a platform data container.\n",
    "\n",
    "Note:\n",
    "- The data-source format for the platform's NoSQL data store is `io.iguaz.v3io.spark.sql.kv`.\n",
    "- The path to the NoSQL table that is associated with the DataFrame should be defined as a fully qualified path of the format `v3io://<data container>/<table path>` &mdash; where `<data container>` is the name of the table's parent data container and `<table path>` is the relative path to the data within the specified container.\n",
    "- You must use the `key` option to define the table's primary key attribute (column). Note that the value of the primary-key attributes must be unique.<br>\n",
    "  You can also optionally use the platform's custom `sorting-key` option to define a sorting-key attribute for the table (which enables performing range scans).<br>\n",
    "  For more information, see the [platform documentation](https://www.iguazio.com/docs/v3.0/data-layer/objects/object-names-and-keys/#sharding-n-sorting-keys)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 2.23 ms, sys: 1.22 ms, total: 3.45 ms\n",
      "Wall time: 942 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Define thepath to your NoSQL table\n",
    "kvStore = os.path.join(file_path+'/stocks_kv')\n",
    "\n",
    "# UUID: key.sorting-key\n",
    "# key: ISIN\n",
    "# sorting-key: Date + Time\n",
    "df.withColumn(\"datetime\", concat(df[\"Date\"], df[\"Time\"]))\\\n",
    "    .write\\\n",
    "    .format(\"io.iguaz.v3io.spark.sql.kv\")\\\n",
    "    .mode(\"append\")\\\n",
    "    .option(\"key\", \"ISIN\")\\\n",
    "    .option(\"sorting-key\", \"datetime\")\\\n",
    "    .option(\"allow-overwrite-schema\", \"true\")\\\n",
    "    .save(kvStore)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"read-data-from-nosql-table-to-spark-df\"></a>\n",
    "#### Load Data a NoSQL Table into a Spark DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2 = spark.read.format(\"io.iguaz.v3io.spark.sql.kv\").load(kvStore)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.43 ms, sys: 3.03 ms, total: 4.46 ms\n",
      "Wall time: 112 ms\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(ISIN='DE000A0WMPJ6', datetime='2018-03-26 00:00:0008:00')]"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df2.select(\"ISIN\", \"datetime\").head(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2.createOrReplaceTempView(\"stock_kv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----------------+\n",
      "|        ISIN|sum(TradedVolume)|\n",
      "+------------+-----------------+\n",
      "|DE0005570808|             3350|\n",
      "|DE000A1K0375|             2000|\n",
      "|DE000A0H08M3|             5394|\n",
      "|LU0488317024|               35|\n",
      "|FI0009000681|            51473|\n",
      "+------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 1.71 ms, sys: 260 µs, total: 1.97 ms\n",
      "Wall time: 459 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "q10 = spark.sql(\"SELECT ISIN, SUM(TradedVolume) FROM stock_kv GROUP BY ISIN\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"write-data-to-partitioned-nosql-table\"></a>\n",
    "#### Persist Data to a Partitioned NoSQL Table\n",
    "\n",
    "Partions are firstly by `Date`, and then by `Time`. <br>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 1.33 ms, sys: 1.09 ms, total: 2.42 ms\n",
      "Wall time: 739 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "# Set Partitioned KV store name\n",
    "kvStorePartition = os.path.join(file_path+'/stocks_kv_partition')\n",
    "\n",
    "# UUID = key.sorting-key\n",
    "# key: ISIN\n",
    "# partition : Date, time\n",
    "df.write\\\n",
    "    .format(\"io.iguaz.v3io.spark.sql.kv\")\\\n",
    "    .mode(\"append\")\\\n",
    "    .option(\"key\", \"ISIN\")\\\n",
    "    .option(\"partition\", \"Date, Time\")\\\n",
    "    .save(kvStorePartition)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"read-data-from-another-nosql-table-to-spark-df\"></a>\n",
    "#### Load Data from Another NoSQL Table into a Spark DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "df3 = spark.read.format(\"io.iguaz.v3io.spark.sql.kv\").load(kvStorePartition)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-------------------+-----+\n",
      "|        ISIN|               Date| Time|\n",
      "+------------+-------------------+-----+\n",
      "|DE0006202005|2018-03-26 00:00:00|08:00|\n",
      "|DE000WCH8881|2018-03-26 00:00:00|08:00|\n",
      "|DE000A11QW68|2018-03-26 00:00:00|08:00|\n",
      "+------------+-------------------+-----+\n",
      "only showing top 3 rows\n",
      "\n",
      "CPU times: user 2.44 ms, sys: 209 µs, total: 2.65 ms\n",
      "Wall time: 152 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "df3.select(\"ISIN\", \"Date\", \"Time\").show(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true
   },
   "source": [
    "<a id=\"test-sql-perf-on-partitioned-nosql-table-w-different-spark-cfgs\"></a>\n",
    "#### Test the SQL Performance on a Partitioned NoSQL Table with Different Spark Configurarions\n",
    "\n",
    "Start out by running the following code to test the performance when using Spark SQL to access a partitioned NoSQL table in the platform with the default [Spark configuration](#initiate-a-spark-session-and-configure-spark).\n",
    "\n",
    "> **Note:** The default Spark configuration doesn't provide good support for the partition data model of the test table, and therefore the test query is expected to hang."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [],
   "source": [
    "df3.createOrReplaceTempView(\"stock_kv_partintion\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+-----------------+\n",
      "|        ISIN|sum(TradedVolume)|\n",
      "+------------+-----------------+\n",
      "|DE000A0H08M3|             5394|\n",
      "|FI0009000681|            51473|\n",
      "|DE000A1K0375|             2000|\n",
      "|DE0005570808|             3350|\n",
      "|LU0488317024|               35|\n",
      "+------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "CPU times: user 2.17 ms, sys: 12 µs, total: 2.19 ms\n",
      "Wall time: 1.07 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 1*m5.2xlarge: Spark 4 executors, 2 cores, and 1 GB per executor; never returns results\n",
    "# 1*m5.2xlarge: Spark 2 executors, 2 cores, and 4 GB per executor; never returns results\n",
    "#%debug\n",
    "q11 = spark.sql(\"SELECT ISIN, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY ISIN\").show(5)\n",
    "#q11 = spark.sql(\"SELECT ISIN, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY ISIN\").explain"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"test-sql-perf-on-partitioned-nosql-table-spark-cfg-experiments\"></a>\n",
    "##### Experiment with Different Spark Configurations\n",
    "\n",
    "The following experiments run the same query with different Spark configurations.\n",
    "To perform the experiments, you need to modify your Spark configuration according to the information in the comments of each job.\n",
    "For information on how to modify the Spark configuration, see the [Modify the Spark Configuration](#modify-the-spark-cfg) section of this tutorial."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+\n",
      "|               Date|sum(TradedVolume)|\n",
      "+-------------------+-----------------+\n",
      "|2018-03-26 00:00:00|         22467641|\n",
      "+-------------------+-----------------+\n",
      "\n",
      "CPU times: user 1.97 ms, sys: 145 µs, total: 2.12 ms\n",
      "Wall time: 814 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 1*m5.2xlarge: Spark 4 executor, 2core and 1G per executor\n",
    "\n",
    "q12 = spark.sql(\"SELECT Date, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY Date\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+\n",
      "|               Date|sum(TradedVolume)|\n",
      "+-------------------+-----------------+\n",
      "|2018-03-26 00:00:00|         22467641|\n",
      "+-------------------+-----------------+\n",
      "\n",
      "CPU times: user 705 µs, sys: 1.88 ms, total: 2.59 ms\n",
      "Wall time: 804 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 1*m5.2xlarge: Spark 2 executor, 2core and 4G per executor,\n",
    "\n",
    "q12 = spark.sql(\"SELECT Date, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY Date\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+\n",
      "|               Date|sum(TradedVolume)|\n",
      "+-------------------+-----------------+\n",
      "|2018-03-26 00:00:00|         22467641|\n",
      "+-------------------+-----------------+\n",
      "\n",
      "CPU times: user 2.48 ms, sys: 0 ns, total: 2.48 ms\n",
      "Wall time: 780 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 1*m5.2xlarge: Spark 4 executor, 1core and 4G per executor,\n",
    "\n",
    "q12 = spark.sql(\"SELECT Date, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY Date\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+\n",
      "|               Date|sum(TradedVolume)|\n",
      "+-------------------+-----------------+\n",
      "|2018-03-26 00:00:00|         22467641|\n",
      "+-------------------+-----------------+\n",
      "\n",
      "CPU times: user 1.92 ms, sys: 132 µs, total: 2.05 ms\n",
      "Wall time: 782 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 1*m5.2xlarge: Spark 3 executor, 1core and 8G per executor,\n",
    "\n",
    "q12 = spark.sql(\"SELECT Date, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY Date\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+\n",
      "|               Date|sum(TradedVolume)|\n",
      "+-------------------+-----------------+\n",
      "|2018-03-26 00:00:00|         22467641|\n",
      "+-------------------+-----------------+\n",
      "\n",
      "CPU times: user 2 ms, sys: 144 µs, total: 2.15 ms\n",
      "Wall time: 822 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 1*m5.2xlarge: Spark 6 executor, 1core and 4G per executor,\n",
    "\n",
    "q12 = spark.sql(\"SELECT Date, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY Date\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+\n",
      "|               Date|sum(TradedVolume)|\n",
      "+-------------------+-----------------+\n",
      "|2018-03-26 00:00:00|         22467641|\n",
      "+-------------------+-----------------+\n",
      "\n",
      "CPU times: user 1.03 ms, sys: 1.04 ms, total: 2.07 ms\n",
      "Wall time: 838 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 1*m5.2xlarge: Spark 3 executor, 1core and 8G per executor,\n",
    "\n",
    "q12 = spark.sql(\"SELECT Date, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY Date\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----------------+\n",
      "|               Date|sum(TradedVolume)|\n",
      "+-------------------+-----------------+\n",
      "|2018-03-26 00:00:00|         22467641|\n",
      "+-------------------+-----------------+\n",
      "\n",
      "CPU times: user 2.41 ms, sys: 47 µs, total: 2.45 ms\n",
      "Wall time: 604 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "# 1*m5.2xlarge: Spark 1 executor, 1core and 20G per executor,\n",
    "\n",
    "q12 = spark.sql(\"SELECT Date, SUM(TradedVolume) FROM stock_kv_partintion GROUP BY Date\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"spark-sql-join\"></a>\n",
    "### Spark SQL Join"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfL = spark.createDataFrame([(\"2504271\", \"LU0378438732\")], [\"SecurityID\", \"ISIN\"])\n",
    "dfR = spark.createDataFrame([(\"2504271\", \"JOIN in Spark SQL\")], [\"SecurityID\", \"SQL Query\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+----------+-----------------+\n",
      "|SecurityID|        ISIN|SecurityID|        SQL Query|\n",
      "+----------+------------+----------+-----------------+\n",
      "|   2504271|LU0378438732|   2504271|JOIN in Spark SQL|\n",
      "+----------+------------+----------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "dfJoin = dfL.join(dfR, dfL.SecurityID == dfR.SecurityID).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfL.createOrReplaceTempView(\"t1\")\n",
    "dfR.createOrReplaceTempView(\"t2\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------+----------+-----------------+\n",
      "|SecurityID|        ISIN|SecurityID|        SQL Query|\n",
      "+----------+------------+----------+-----------------+\n",
      "|   2504271|LU0378438732|   2504271|JOIN in Spark SQL|\n",
      "+----------+------------+----------+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "qJoin = spark.sql(\"SELECT * FROM t1, t2 where t1.SecurityID=t2.SecurityID\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"spark-sql-on-parquet\"></a>\n",
    "### Spark SQL on a Parquet File"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Persist Data into Iguazio Data Container in Parquet format\n",
    "\n",
    "Use the same stock dataset to store in Parquet format."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 756 µs, sys: 1.16 ms, total: 1.92 ms\n",
      "Wall time: 909 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "parqFile = os.path.join(file_path+'/stocks_parq')\n",
    "\n",
    "df.write\\\n",
    "    .mode(\"overwrite\")\\\n",
    "    .parquet(parqFile)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "dfPARQ = spark.read.parquet(parqFile)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(ISIN='CH0038389992', Date=datetime.datetime(2018, 3, 26, 0, 0))"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dfPARQ.select(\"ISIN\", \"Date\").head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true
   },
   "source": [
    "<a id=\"spark-sql-on-partitioned-table\"></a>\n",
    "### Spark SQL on a Partitioned Table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Create a Partitioned Table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This examples creates a partitioned \"weather\" table.  The `option(\"partition\", \"year, month, day\")` write option partitions the table by the year, month, and day item attributes. As demonstrated in the following image, if you browse the container in the dashboard after running the example, you'll see that the weather directory has **year=&lt;value&gt;/month=&lt;value&gt;/day=&lt;value&gt;** partition directories that match the written items. If you select any of the nested day partition directories, you can see the written items and their attributes. For example, the first item (with attribute values 2016, 3, 25, 6, 16, 0.00, 55) is saved to a 20163256 file in a **weather/year=2016/month=3/day=25** partition directory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [],
   "source": [
    "table_path = os.path.join(os.getenv('V3IO_HOME_URL')+'/examples/weather/')\n",
    "\n",
    "df = spark.createDataFrame([\n",
    "    (2016,  3, 25, 17, 18, 0.2, 62),\n",
    "    (2016,  7, 24,  7, 19, 0.0, 52),\n",
    "    (2016, 12, 24,  9, 10, 0.1, 47),\n",
    "    (2017,  5,  7, 14, 21, 0.0, 70),\n",
    "    (2017, 11,  1, 10, 15, 0.0, 34),\n",
    "    (2017, 12, 12, 16, 12, 0.0, 47),\n",
    "    (2017, 12, 24, 17, 11, 1.0, 50),\n",
    "    (2018,  1, 18, 17, 10, 2.0, 45),\n",
    "    (2018,  5, 20, 21, 20, 0.0, 59),\n",
    "    (2018, 11,  1, 11, 11, 0.1, 65)\n",
    "], [\"year\", \"month\", \"day\", \"hour\", \"degrees_cel\", \"rain_ml\", \"humidity_per\"])\n",
    "\n",
    "df_with_key = df.withColumn(\n",
    "    \"time\", concat(df[\"year\"], df[\"month\"], df[\"day\"], df[\"hour\"]))\n",
    "\n",
    "df_with_key.write.format(\"io.iguaz.v3io.spark.sql.kv\") \\\n",
    "    .mode(\"overwrite\") \\\n",
    "    .option(\"key\", \"time\") \\\n",
    "    .option(\"partition\", \"year, month, day, hour\") \\\n",
    "    .save(table_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Reading from partition table\n",
    "\n",
    "Following is the output of the example's show commands for each read. The filtered results are gathered by scanning only the partition directories that match the filter criteria."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Perform A Full Table Scan"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "|year|month|day|hour|degrees_cel|rain_ml|humidity_per|      time|\n",
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "|2016|   12| 24|   9|         10|    0.1|          47| 201612249|\n",
      "|2016|    3| 25|  17|         18|    0.2|          62| 201632517|\n",
      "|2016|    7| 24|   7|         19|    0.0|          52|  20167247|\n",
      "|2017|   11|  1|  10|         15|    0.0|          34| 201711110|\n",
      "|2017|   12| 12|  16|         12|    0.0|          47|2017121216|\n",
      "|2017|   12| 24|  17|         11|    1.0|          50|2017122417|\n",
      "|2017|    5|  7|  14|         21|    0.0|          70|  20175714|\n",
      "|2018|    1| 18|  17|         10|    2.0|          45| 201811817|\n",
      "|2018|   11|  1|  11|         11|    0.1|          65| 201811111|\n",
      "|2018|    5| 20|  21|         20|    0.0|          59| 201852021|\n",
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "readDF = spark.read.format(\"io.iguaz.v3io.spark.sql.kv\").load(table_path)\n",
    "readDF.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Retrieve all data in the last six months of each year:\n",
    "\n",
    "Filter: month > 6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "|year|month|day|hour|degrees_cel|rain_ml|humidity_per|      time|\n",
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "|2016|   12| 24|   9|         10|    0.1|          47| 201612249|\n",
      "|2016|    7| 24|   7|         19|    0.0|          52|  20167247|\n",
      "|2017|   11|  1|  10|         15|    0.0|          34| 201711110|\n",
      "|2017|   12| 12|  16|         12|    0.0|          47|2017121216|\n",
      "|2017|   12| 24|  17|         11|    1.0|          50|2017122417|\n",
      "|2018|   11|  1|  11|         11|    0.1|          65| 201811111|\n",
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "readDF = spark.read.format(\"io.iguaz.v3io.spark.sql.kv\").load(table_path) \\\n",
    "    .filter(\"month > 6\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Retrieve all hours in Dec 24 of each year:\n",
    "\n",
    "Filter: month == 12 AND day == 24 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "|year|month|day|hour|degrees_cel|rain_ml|humidity_per|      time|\n",
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "|2016|   12| 24|   9|         10|    0.1|          47| 201612249|\n",
      "|2017|   12| 24|  17|         11|    1.0|          50|2017122417|\n",
      "+----+-----+---+----+-----------+-------+------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "readDF = spark.read.format(\"io.iguaz.v3io.spark.sql.kv\").load(table_path) \\\n",
    "    .filter(\"month == 12 AND day == 24\") \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Retrieve data during 08:00&ndash;20:00 each day in the last six months of each year"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+---+----+-----------+-------+------------+---------+\n",
      "|year|month|day|hour|degrees_cel|rain_ml|humidity_per|     time|\n",
      "+----+-----+---+----+-----------+-------+------------+---------+\n",
      "|2016|    3| 25|  17|         18|    0.2|          62|201632517|\n",
      "|2017|    5|  7|  14|         21|    0.0|          70| 20175714|\n",
      "|2018|    1| 18|  17|         10|    2.0|          45|201811817|\n",
      "+----+-----+---+----+-----------+-------+------------+---------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "readDF = spark.read.format(\"io.iguaz.v3io.spark.sql.kv\").load(table_path) \\\n",
    "    .filter(\"month < 7 AND hour >= 8 AND hour <= 20\") \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true
   },
   "source": [
    "<a id=\"conditional-update\"></a>\n",
    "## Perform Conditional Data Updates\n",
    "\n",
    "This example demonstrates how to conditionally update NoSQL table items by using a conditional write option.\n",
    "Each `write` command in the example is followed by matching `read` and `show` commands to read and display the value of the updated item in the target table after the write operation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"conditional-update-generate-data\"></a>\n",
    "### Generate Data\n",
    "\n",
    "The first write command writes an item (row) to a \"cars\" table. The item's `reg_license` primary-key (identity-column) attribute is set to 7843321, the mode attribute is set to \"Honda\", and the odometer attribute is set to `29321`. The `overwrite` save mode is used to overwrite the table if it already exists and create it otherwise. Reading the item from the table produces this output:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+--------+\n",
      "|reg_license|model|odometer|\n",
      "+-----------+-----+--------+\n",
      "|    7843321|Honda|   29321|\n",
      "+-----------+-----+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "writeDF = spark.createDataFrame([(\"7843321\", \"Honda\", 29321)],\n",
    "                                [\"reg_license\", \"model\", \"odometer\"])\n",
    "\n",
    "writeDF.write.format(\"io.iguaz.v3io.spark.sql.kv\") \\\n",
    "    .option(\"key\", \"reg_license\") \\\n",
    "    .mode(\"overwrite\") \\\n",
    "    .save(os.path.join(os.getenv('V3IO_HOME_URL'))+'/cars/')\n",
    "\n",
    "readDF = spark.read.format(\"io.iguaz.v3io.spark.sql.kv\") \\\n",
    "    .load(os.path.join(os.getenv('V3IO_HOME_URL'))+'/cars/') \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"conditional-update-perform-update\"></a>\n",
    "### Conditionally Update the Data\n",
    "\n",
    "Update the odometer to `31718` on the condition that the new odometer value is greater than the old value.\n",
    "This ensures that the `odometer` attribute (column) reflects the most updated value of the odometer."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-----+--------+\n",
      "|reg_license|model|odometer|\n",
      "+-----------+-----+--------+\n",
      "|    7843321|Honda|   31718|\n",
      "+-----------+-----+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "writeDF = spark.createDataFrame([(\"7843321\", \"Honda\", 31718)],\n",
    "                                [\"reg_license\", \"model\", \"odometer\"])\n",
    "\n",
    "writeDF.write.format(\"io.iguaz.v3io.spark.sql.kv\") \\\n",
    "    .option(\"key\", \"reg_license\") \\\n",
    "    .option(\"condition\", \"${odometer} > odometer\") \\\n",
    "    .mode(\"append\") \\\n",
    "    .save(os.path.join(os.getenv('V3IO_HOME_URL'))+'/cars/')\n",
    "\n",
    "readDF = spark.read.format(\"io.iguaz.v3io.spark.sql.kv\") \\\n",
    "    .load(os.path.join(os.getenv('V3IO_HOME_URL'))+'/cars/') \\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<font color=green> **Congratulations!**</font> You've completed the Spark SQL Analytics with the Iguazio Data Science Platform tutorial."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": true,
    "toc-nb-collapsed": true
   },
   "source": [
    "## Cleanup\n",
    "\n",
    "Prior to exiting, let's do housekeeping to release disk space, computation and memory resources taken by this session."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Remove Data\n",
    "\n",
    "When you are done, uncomment the remove command in the following code to remove the example directory:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Uncomment the following line to remove the examples directory:\n",
    "# rm -rf /v3io/${V3IO_HOME}/examples/*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "toc-hr-collapsed": false
   },
   "source": [
    "<a id=\"stop-spark-session\"></a>\n",
    "### Stop the Spark Session\n",
    "\n",
    "Run the following command to release the computation and memory resources that are being consumed by your Spark session:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}