jupyter/notebook/batch_etl_steps.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "\n",
    "findspark.add_jars('/app/postgresql-42.1.4.jar')\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = (\n",
    "    SparkSession.builder\n",
    "    .appName(\"Stocks:ETL\")\n",
    "    .config(\"spark.driver.memory\", \"512m\")\n",
    "    .config(\"spark.driver.cores\", \"1\")\n",
    "    .config(\"spark.executor.memory\", \"512m\")\n",
    "    .config(\"spark.executor.cores\", \"1\")\n",
    "    .config(\"spark.sql.shuffle.partitions\", \"2\")\n",
    "    .getOrCreate()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'2.4.5'"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.version"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "stocks_dir = '/dataset/stocks-small'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# UDF\n",
    "from pyspark.sql.types import StringType\n",
    "#\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.window import Window"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read \\\n",
    "    .option(\"header\", True) \\\n",
    "    .option(\"inferSchema\", True) \\\n",
    "    .csv(stocks_dir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Date: timestamp (nullable = true)\n",
      " |-- Open: double (nullable = true)\n",
      " |-- High: double (nullable = true)\n",
      " |-- Low: double (nullable = true)\n",
      " |-- Close: double (nullable = true)\n",
      " |-- Volume: integer (nullable = true)\n",
      " |-- OpenInt: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.count()\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+------+-------+\n",
      "|               Date|  Open|  High|   Low| Close|Volume|OpenInt|\n",
      "+-------------------+------+------+------+------+------+-------+\n",
      "|1962-01-02 00:00:00| 6.413| 6.413|6.3378|6.3378|467056|      0|\n",
      "|1962-01-03 00:00:00|6.3378|6.3963|6.3378|6.3963|350294|      0|\n",
      "|1962-01-04 00:00:00|6.3963|6.3963|6.3295|6.3295|314365|      0|\n",
      "|1962-01-05 00:00:00|6.3211|6.3211|6.1958|6.2041|440112|      0|\n",
      "|1962-01-08 00:00:00|6.2041|6.2041|6.0373| 6.087|655676|      0|\n",
      "|1962-01-09 00:00:00|6.1208|6.2376|6.1208|6.1621|592806|      0|\n",
      "|1962-01-10 00:00:00|6.1707|6.2041|6.1707|6.1707|359274|      0|\n",
      "|1962-01-11 00:00:00|6.1875|6.2376|6.1875|6.2376|386220|      0|\n",
      "|1962-01-12 00:00:00|6.2543|6.2962|6.2543|6.2543|529933|      0|\n",
      "|1962-01-15 00:00:00|6.2708|6.2962|6.2708|6.2792|305383|      0|\n",
      "|1962-01-16 00:00:00|6.2708|6.2708|6.2128|6.2128|305383|      0|\n",
      "|1962-01-17 00:00:00|6.1875|6.1875|6.0956|6.1125|502984|      0|\n",
      "|1962-01-18 00:00:00|6.1291|6.1875|6.1291|6.1291|449093|      0|\n",
      "|1962-01-19 00:00:00|6.1291|6.1457|6.0624|6.1374|485021|      0|\n",
      "|1962-01-22 00:00:00|6.1374|6.1958|6.1208|6.1208|332329|      0|\n",
      "|1962-01-23 00:00:00|6.1208|6.1291|6.0538|6.0624|449093|      0|\n",
      "|1962-01-24 00:00:00|6.0624|6.0956|6.0287|6.0956|494001|      0|\n",
      "|1962-01-25 00:00:00|6.0956|6.1457|6.0208|6.0287|386220|      0|\n",
      "|1962-01-26 00:00:00|6.0287|6.0538|5.9951|5.9951|296401|      0|\n",
      "|1962-01-29 00:00:00|5.9951|6.0373|5.8952|5.8952|700585|      0|\n",
      "+-------------------+------+------+------+------+------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumn('filename', F.input_file_name())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+------+-------+---------------------------------------+\n",
      "|Date               |Open  |High  |Low   |Close |Volume|OpenInt|filename                               |\n",
      "+-------------------+------+------+------+------+------+-------+---------------------------------------+\n",
      "|1962-01-02 00:00:00|6.413 |6.413 |6.3378|6.3378|467056|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-03 00:00:00|6.3378|6.3963|6.3378|6.3963|350294|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-04 00:00:00|6.3963|6.3963|6.3295|6.3295|314365|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-05 00:00:00|6.3211|6.3211|6.1958|6.2041|440112|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-08 00:00:00|6.2041|6.2041|6.0373|6.087 |655676|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-09 00:00:00|6.1208|6.2376|6.1208|6.1621|592806|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-10 00:00:00|6.1707|6.2041|6.1707|6.1707|359274|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-11 00:00:00|6.1875|6.2376|6.1875|6.2376|386220|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-12 00:00:00|6.2543|6.2962|6.2543|6.2543|529933|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-15 00:00:00|6.2708|6.2962|6.2708|6.2792|305383|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-16 00:00:00|6.2708|6.2708|6.2128|6.2128|305383|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-17 00:00:00|6.1875|6.1875|6.0956|6.1125|502984|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-18 00:00:00|6.1291|6.1875|6.1291|6.1291|449093|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-19 00:00:00|6.1291|6.1457|6.0624|6.1374|485021|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-22 00:00:00|6.1374|6.1958|6.1208|6.1208|332329|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-23 00:00:00|6.1208|6.1291|6.0538|6.0624|449093|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-24 00:00:00|6.0624|6.0956|6.0287|6.0956|494001|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-25 00:00:00|6.0956|6.1457|6.0208|6.0287|386220|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-26 00:00:00|6.0287|6.0538|5.9951|5.9951|296401|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "|1962-01-29 00:00:00|5.9951|6.0373|5.8952|5.8952|700585|0      |file:///dataset/stocks-small/ibm.us.txt|\n",
      "+-------------------+------+------+------+------+------+-------+---------------------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_lookup = spark.read.csv('/dataset/yahoo-symbols-201709.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+--------+--------------------+-------+\n",
      "|   _c0|                 _c1|     _c2|                 _c3|    _c4|\n",
      "+------+--------------------+--------+--------------------+-------+\n",
      "|Ticker|                Name|Exchange|       Category Name|Country|\n",
      "|  OEDV|Osage Exploration...|     PNK|                null|    USA|\n",
      "|  AAPL|          Apple Inc.|     NMS|Electronic Equipment|    USA|\n",
      "|   BAC|Bank of America C...|     NYQ|  Money Center Banks|    USA|\n",
      "|  AMZN|    Amazon.com, Inc.|     NMS|Catalog & Mail Or...|    USA|\n",
      "|     T|           AT&T Inc.|     NYQ|Telecom Services ...|    USA|\n",
      "|  GOOG|       Alphabet Inc.|     NMS|Internet Informat...|    USA|\n",
      "|    MO|  Altria Group, Inc.|     NYQ|          Cigarettes|    USA|\n",
      "|   DAL|Delta Air Lines, ...|     NYQ|      Major Airlines|    USA|\n",
      "|    AA|   Alcoa Corporation|     NYQ|            Aluminum|    USA|\n",
      "|   AXP|American Express ...|     NYQ|     Credit Services|    USA|\n",
      "|    DD|E. I. du Pont de ...|     NYQ|Agricultural Chem...|    USA|\n",
      "|  BABA|Alibaba Group Hol...|     NYQ|Specialty Retail,...|    USA|\n",
      "|   ABT| Abbott Laboratories|     NYQ|Medical Appliance...|    USA|\n",
      "|    UA|  Under Armour, Inc.|     NYQ|Textile - Apparel...|    USA|\n",
      "|  AMAT|Applied Materials...|     NMS|Semiconductor Equ...|    USA|\n",
      "|  AMGN|          Amgen Inc.|     NMS|       Biotechnology|    USA|\n",
      "|   AAL|American Airlines...|     NMS|      Major Airlines|    USA|\n",
      "|   AIG|American Internat...|     NYQ|Property & Casual...|    USA|\n",
      "|   ALL|The Allstate Corp...|     NYQ|Property & Casual...|    USA|\n",
      "+------+--------------------+--------+--------------------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_lookup.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_symbol_from(filename):\n",
    "    return filename.split('/')[-1].split('.')[0].upper()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'IBM'"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# filename = 'file:///dataset/stocks-small/ibm.us.txt' # => IBM\n",
    "extract_symbol_from('file:///dataset/stocks-small/ibm.us.txt')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "extract_symbol = F.udf(lambda filename: extract_symbol_from(filename), StringType())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "stocks_folder = stocks_dir\n",
    "df = spark.read \\\n",
    "        .option(\"header\", True) \\\n",
    "        .option(\"inferSchema\", True) \\\n",
    "        .csv(stocks_folder) \\\n",
    "        .withColumn(\"name\", extract_symbol(F.input_file_name()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+------+-------+----+\n",
      "|               Date|  Open|  High|   Low| Close|Volume|OpenInt|name|\n",
      "+-------------------+------+------+------+------+------+-------+----+\n",
      "|1962-01-02 00:00:00| 6.413| 6.413|6.3378|6.3378|467056|      0| IBM|\n",
      "|1962-01-03 00:00:00|6.3378|6.3963|6.3378|6.3963|350294|      0| IBM|\n",
      "|1962-01-04 00:00:00|6.3963|6.3963|6.3295|6.3295|314365|      0| IBM|\n",
      "|1962-01-05 00:00:00|6.3211|6.3211|6.1958|6.2041|440112|      0| IBM|\n",
      "|1962-01-08 00:00:00|6.2041|6.2041|6.0373| 6.087|655676|      0| IBM|\n",
      "+-------------------+------+------+------+------+------+-------+----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read \\\n",
    "        .option(\"header\", True) \\\n",
    "        .option(\"inferSchema\", True) \\\n",
    "        .csv(stocks_folder) \\\n",
    "        .withColumn(\"name\", extract_symbol(F.input_file_name())) \\\n",
    "        .withColumnRenamed(\"Date\", \"dateTime\") \\\n",
    "        .withColumnRenamed(\"Open\", \"open\") \\\n",
    "        .withColumnRenamed(\"High\", \"high\") \\\n",
    "        .withColumnRenamed(\"Low\", \"low\") \\\n",
    "        .withColumnRenamed(\"Close\", \"close\") \\\n",
    "        .drop(\"Volume\", \"OpenInt\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_stocks = df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+----+\n",
      "|           dateTime|  open|  high|   low| close|name|\n",
      "+-------------------+------+------+------+------+----+\n",
      "|1962-01-02 00:00:00| 6.413| 6.413|6.3378|6.3378| IBM|\n",
      "|1962-01-03 00:00:00|6.3378|6.3963|6.3378|6.3963| IBM|\n",
      "|1962-01-04 00:00:00|6.3963|6.3963|6.3295|6.3295| IBM|\n",
      "|1962-01-05 00:00:00|6.3211|6.3211|6.1958|6.2041| IBM|\n",
      "|1962-01-08 00:00:00|6.2041|6.2041|6.0373| 6.087| IBM|\n",
      "+-------------------+------+------+------+------+----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_stocks.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "lookup_file = '/dataset/yahoo-symbols-201709.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "symbols_lookup = spark.read. \\\n",
    "        option(\"header\", True). \\\n",
    "        option(\"inferSchema\", True). \\\n",
    "        csv(lookup_file). \\\n",
    "        select(\"Ticker\", \"Category Name\"). \\\n",
    "        withColumnRenamed(\"Ticker\", \"symbol\"). \\\n",
    "        withColumnRenamed(\"Category Name\", \"category\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------+------+------+------+----+\n",
      "|           dateTime|  open|  high|   low| close|name|\n",
      "+-------------------+------+------+------+------+----+\n",
      "|1962-01-02 00:00:00| 6.413| 6.413|6.3378|6.3378| IBM|\n",
      "|1962-01-03 00:00:00|6.3378|6.3963|6.3378|6.3963| IBM|\n",
      "|1962-01-04 00:00:00|6.3963|6.3963|6.3295|6.3295| IBM|\n",
      "+-------------------+------+------+------+------+----+\n",
      "only showing top 3 rows\n",
      "\n",
      "+------+--------------------+\n",
      "|symbol|            category|\n",
      "+------+--------------------+\n",
      "|  OEDV|                null|\n",
      "|  AAPL|Electronic Equipment|\n",
      "|   BAC|  Money Center Banks|\n",
      "+------+--------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_stocks.show(3)\n",
    "symbols_lookup.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "joined_df = df_stocks \\\n",
    "    .withColumnRenamed('dateTime', \"full_date\") \\\n",
    "    .filter(\"full_date >= \\\"2017-09-01\\\"\") \\\n",
    "    .withColumn(\"year\", F.year(\"full_date\")) \\\n",
    "    .withColumn(\"month\", F.month(\"full_date\")) \\\n",
    "    .withColumn(\"day\", F.dayofmonth(\"full_date\")) \\\n",
    "    .withColumnRenamed(\"name\", \"symbol\") \\\n",
    "    .join(symbols_lookup, [\"symbol\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------------------+------+------+------+------+----+-----+---+--------------------+\n",
      "|symbol|          full_date|  open|  high|   low| close|year|month|day|            category|\n",
      "+------+-------------------+------+------+------+------+----+-----+---+--------------------+\n",
      "|   IBM|2017-01-03 00:00:00|160.76| 161.6|159.81|160.95|2017|    1|  3|Information Techn...|\n",
      "|   IBM|2017-01-04 00:00:00|161.51|163.53|161.11|162.94|2017|    1|  4|Information Techn...|\n",
      "|   IBM|2017-01-05 00:00:00|162.93|163.06|161.01|162.41|2017|    1|  5|Information Techn...|\n",
      "+------+-------------------+------+------+------+------+----+-----+---+--------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joined_df.show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "window20 = (Window.partitionBy(F.col('symbol')).orderBy(F.col(\"full_date\")).rowsBetween(-20, 0))\n",
    "window50 = (Window.partitionBy(F.col('symbol')).orderBy(F.col(\"full_date\")).rowsBetween(-50, 0))\n",
    "window100 = (Window.partitionBy(F.col('symbol')).orderBy(F.col(\"full_date\")).rowsBetween(-100, 0))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "stocks_moving_avg_df = joined_df \\\n",
    "    .withColumn(\"ma20\", F.avg(\"close\").over(window20)) \\\n",
    "    .withColumn(\"ma50\", F.avg(\"close\").over(window50)) \\\n",
    "    .withColumn(\"ma100\", F.avg(\"close\").over(window100))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+------------------+\n",
      "|symbol| close|              ma20|\n",
      "+------+------+------------------+\n",
      "|  AAPL|114.31|            114.31|\n",
      "|  AAPL|114.19|            114.25|\n",
      "|  AAPL|114.77|114.42333333333333|\n",
      "|  AAPL|116.04|          114.8275|\n",
      "|  AAPL|117.11|115.28399999999999|\n",
      "|  AAPL|117.23|115.60833333333333|\n",
      "|  AAPL|117.86|115.92999999999999|\n",
      "|  AAPL|117.37|            116.11|\n",
      "|  AAPL|117.16|116.22666666666666|\n",
      "|  AAPL| 118.1|116.41399999999999|\n",
      "|  AAPL|118.09|116.56636363636362|\n",
      "|  AAPL|117.89|116.67666666666666|\n",
      "|  AAPL| 118.1|116.78615384615384|\n",
      "|  AAPL|118.19|116.88642857142857|\n",
      "|  AAPL|118.07|116.96533333333332|\n",
      "|  AAPL|119.95|117.15187499999999|\n",
      "|  AAPL|120.01|            117.32|\n",
      "|  AAPL|120.02|            117.47|\n",
      "|  AAPL| 119.7|117.58736842105263|\n",
      "|  AAPL|119.43|117.67949999999999|\n",
      "|  AAPL| 126.7| 118.1090476190476|\n",
      "|  AAPL| 126.5|118.68952380952379|\n",
      "|  AAPL|127.03| 119.3009523809524|\n",
      "|  AAPL|128.23|119.94190476190477|\n",
      "|  AAPL|129.44|120.58000000000004|\n",
      "+------+------+------------------+\n",
      "only showing top 25 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Moving Average\n",
    "stocks_moving_avg_df.select('symbol', 'close', 'ma20').show(25)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "output_dir = '/dataset/output.parquet'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "stocks_moving_avg_df \\\n",
    "    .write \\\n",
    "    .mode('overwrite') \\\n",
    "    .partitionBy(\"year\", \"month\", \"day\") \\\n",
    "    .parquet(output_dir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_parquet = spark.read.parquet(output_dir)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4142"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_parquet.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_parquet.createOrReplaceTempView(\"stocks\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(2) HashAggregate(keys=[symbol#559], functions=[max(close#564)])\n",
      "+- Exchange hashpartitioning(symbol#559, 200)\n",
      "   +- *(1) HashAggregate(keys=[symbol#559], functions=[partial_max(close#564)])\n",
      "      +- *(1) Project [symbol#559, close#564]\n",
      "         +- *(1) Filter ((isnotnull(full_date#560) && (cast(full_date#560 as string) >= 2017-09-01)) && (cast(full_date#560 as string) < 2017-10-01))\n",
      "            +- *(1) FileScan parquet [symbol#559,full_date#560,close#564,year#569,month#570,day#571] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/dataset/output.parquet], PartitionCount: 218, PartitionFilters: [], PushedFilters: [IsNotNull(full_date)], ReadSchema: struct<symbol:string,full_date:timestamp,close:double>\n"
     ]
    }
   ],
   "source": [
    "badHighestClosingPrice = spark.sql(\"SELECT symbol, MAX(close) AS price FROM stocks WHERE full_date >= '2017-09-01' AND full_date < '2017-10-01' GROUP BY symbol\")\n",
    "badHighestClosingPrice.explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(2) HashAggregate(keys=[symbol#559], functions=[max(close#564)])\n",
      "+- Exchange hashpartitioning(symbol#559, 200)\n",
      "   +- *(1) HashAggregate(keys=[symbol#559], functions=[partial_max(close#564)])\n",
      "      +- *(1) Project [symbol#559, close#564]\n",
      "         +- *(1) FileScan parquet [symbol#559,close#564,year#569,month#570,day#571] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/dataset/output.parquet], PartitionCount: 20, PartitionFilters: [isnotnull(year#569), isnotnull(month#570), (year#569 = 2017), (month#570 = 9)], PushedFilters: [], ReadSchema: struct<symbol:string,close:double>\n"
     ]
    }
   ],
   "source": [
    "highestClosingPrice = spark.sql(\"SELECT symbol, MAX(close) AS price FROM stocks WHERE year=2017 AND month=9 GROUP BY symbol\")\n",
    "highestClosingPrice.explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Write to Postgres\n",
    "stocks_moving_avg_df \\\n",
    "    .drop(\"year\", \"month\", \"day\") \\\n",
    "    .write \\\n",
    "    .format(\"jdbc\") \\\n",
    "    .option(\"url\", \"jdbc:postgresql://postgres/workshop\") \\\n",
    "    .option(\"dbtable\", \"workshop.stocks\") \\\n",
    "    .option(\"user\", \"workshop\") \\\n",
    "    .option(\"password\", \"w0rkzh0p\") \\\n",
    "    .option(\"driver\", \"org.postgresql.Driver\") \\\n",
    "    .mode('append') \\\n",
    "    .save()"
   ]
  },
  {
   "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}