SparkSQL.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Hello"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## This is my first spark project"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Hola mundo\n"
     ]
    }
   ],
   "source": [
    "print(\"Hola mundo\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "2 + 4"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "\n",
    "sc = SparkContext.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.context.SparkContext"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = [1, 2, 3, 4, 5]\n",
    "distData = sc.parallelize(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(name='Alice', age=1)]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "l = [('Alice', 1)]\n",
    "df = sqlContext.createDataFrame(l, ['name', 'age'])\n",
    "df.collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['Date,Open,High,Low,Close,AdjClose,Volume',\n",
       " '2019-07-15,248.000000,254.419998,244.860001,253.500000,253.500000,11000100',\n",
       " '2019-07-16,249.300003,253.529999,247.929993,252.380005,252.380005,8149000',\n",
       " '2019-07-17,255.669998,258.309998,253.350006,254.860001,254.860001,9764700',\n",
       " '2019-07-18,255.050003,255.750000,251.889999,253.539993,253.539993,4764500']"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data_file = 'TSLA.csv'\n",
    "raw_rdd = sc.textFile(data_file).cache()\n",
    "raw_rdd.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.RDD"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(raw_rdd)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[['Date', 'Open', 'High', 'Low', 'Close', 'AdjClose', 'Volume'], ['2019-07-15', '248.000000', '254.419998', '244.860001', '253.500000', '253.500000', '11000100']]\n",
      "<class 'pyspark.rdd.PipelinedRDD'>\n"
     ]
    }
   ],
   "source": [
    "csv_rdd = raw_rdd.map(lambda row: row.split(\",\"))\n",
    "print(csv_rdd.take(2))\n",
    "print(type(csv_rdd))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "7"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "len(csv_rdd.take(1)[0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = sqlContext.read.load(data_file, \n",
    "                      format='com.databricks.spark.csv', \n",
    "                      header='true', \n",
    "                      inferSchema='true')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "253"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(Date='2019-07-15', Open=248.0, High=254.419998, Low=244.860001, Close=253.5, AdjClose=253.5, Volume=11000100),\n",
       " Row(Date='2019-07-16', Open=249.300003, High=253.529999, Low=247.929993, Close=252.380005, AdjClose=252.380005, Volume=8149000),\n",
       " Row(Date='2019-07-17', Open=255.669998, High=258.309998, Low=253.350006, Close=254.860001, AdjClose=254.860001, Volume=9764700),\n",
       " Row(Date='2019-07-18', Open=255.050003, High=255.75, Low=251.889999, Close=253.539993, AdjClose=253.539993, Volume=4764500),\n",
       " Row(Date='2019-07-19', Open=255.690002, High=259.959991, Low=254.619995, Close=258.179993, AdjClose=258.179993, Volume=7048400)]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Date: string (nullable = true)\n",
      " |-- Open: double (nullable = true)\n",
      " |-- High: double (nullable = true)\n",
      " |-- Low: double (nullable = true)\n",
      " |-- Close: double (nullable = true)\n",
      " |-- AdjClose: double (nullable = true)\n",
      " |-- Volume: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "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>Date</th>\n",
       "      <th>Open</th>\n",
       "      <th>High</th>\n",
       "      <th>Low</th>\n",
       "      <th>Close</th>\n",
       "      <th>AdjClose</th>\n",
       "      <th>Volume</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2019-07-15</td>\n",
       "      <td>248.000000</td>\n",
       "      <td>254.419998</td>\n",
       "      <td>244.860001</td>\n",
       "      <td>253.500000</td>\n",
       "      <td>253.500000</td>\n",
       "      <td>11000100</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2019-07-16</td>\n",
       "      <td>249.300003</td>\n",
       "      <td>253.529999</td>\n",
       "      <td>247.929993</td>\n",
       "      <td>252.380005</td>\n",
       "      <td>252.380005</td>\n",
       "      <td>8149000</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2019-07-17</td>\n",
       "      <td>255.669998</td>\n",
       "      <td>258.309998</td>\n",
       "      <td>253.350006</td>\n",
       "      <td>254.860001</td>\n",
       "      <td>254.860001</td>\n",
       "      <td>9764700</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2019-07-18</td>\n",
       "      <td>255.050003</td>\n",
       "      <td>255.750000</td>\n",
       "      <td>251.889999</td>\n",
       "      <td>253.539993</td>\n",
       "      <td>253.539993</td>\n",
       "      <td>4764500</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2019-07-19</td>\n",
       "      <td>255.690002</td>\n",
       "      <td>259.959991</td>\n",
       "      <td>254.619995</td>\n",
       "      <td>258.179993</td>\n",
       "      <td>258.179993</td>\n",
       "      <td>7048400</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "         Date        Open        High         Low       Close    AdjClose  \\\n",
       "0  2019-07-15  248.000000  254.419998  244.860001  253.500000  253.500000   \n",
       "1  2019-07-16  249.300003  253.529999  247.929993  252.380005  252.380005   \n",
       "2  2019-07-17  255.669998  258.309998  253.350006  254.860001  254.860001   \n",
       "3  2019-07-18  255.050003  255.750000  251.889999  253.539993  253.539993   \n",
       "4  2019-07-19  255.690002  259.959991  254.619995  258.179993  258.179993   \n",
       "\n",
       "     Volume  \n",
       "0  11000100  \n",
       "1   8149000  \n",
       "2   9764700  \n",
       "3   4764500  \n",
       "4   7048400  "
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pandas\n",
    "df.toPandas().head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *\n",
    "schema = StructType([\n",
    "   StructField(\"date\", StringType(), True),\n",
    "   StructField(\"openprice\", IntegerType(), True),\n",
    "   StructField(\"highprice\", IntegerType(), True),\n",
    "   StructField(\"lowprice\", IntegerType(), True),\n",
    "   StructField(\"closeprice\", IntegerType(), True),\n",
    "   StructField(\"volume\", IntegerType(), True),\n",
    "   StructField(\"adjcloseprice\", IntegerType(), True)])\n",
    "\n",
    "\n",
    "df2 = sqlContext.read.load(data_file, \n",
    "                      format='com.databricks.spark.csv', \n",
    "                      header='true', \n",
    "                      schema=schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(date='2019-07-15', openprice=None, highprice=None, lowprice=None, closeprice=None, volume=None, adjcloseprice=11000100),\n",
       " Row(date='2019-07-16', openprice=None, highprice=None, lowprice=None, closeprice=None, volume=None, adjcloseprice=8149000)]"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df2.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- date: string (nullable = true)\n",
      " |-- openprice: integer (nullable = true)\n",
      " |-- highprice: integer (nullable = true)\n",
      " |-- lowprice: integer (nullable = true)\n",
      " |-- closeprice: integer (nullable = true)\n",
      " |-- volume: integer (nullable = true)\n",
      " |-- adjcloseprice: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df2.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.dataframe.DataFrame"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+----------+----------+----------+----------+--------+\n",
      "|      Date|      Open|      High|       Low|     Close|  AdjClose|  Volume|\n",
      "+----------+----------+----------+----------+----------+----------+--------+\n",
      "|2019-07-15|     248.0|254.419998|244.860001|     253.5|     253.5|11000100|\n",
      "|2019-07-16|249.300003|253.529999|247.929993|252.380005|252.380005| 8149000|\n",
      "|2019-07-17|255.669998|258.309998|253.350006|254.860001|254.860001| 9764700|\n",
      "|2019-07-18|255.050003|    255.75|251.889999|253.539993|253.539993| 4764500|\n",
      "|2019-07-19|255.690002|259.959991|254.619995|258.179993|258.179993| 7048400|\n",
      "+----------+----------+----------+----------+----------+----------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+----------+----------+----------+----------+--------+\n",
      "|      Date|      Open|      High|       Low|     Close|  AdjClose|  Volume|\n",
      "+----------+----------+----------+----------+----------+----------+--------+\n",
      "|2019-07-15|     248.0|254.419998|244.860001|     253.5|     253.5|11000100|\n",
      "|2019-07-16|249.300003|253.529999|247.929993|252.380005|252.380005| 8149000|\n",
      "|2019-07-17|255.669998|258.309998|253.350006|254.860001|254.860001| 9764700|\n",
      "|2019-07-18|255.050003|    255.75|251.889999|253.539993|253.539993| 4764500|\n",
      "|2019-07-19|255.690002|259.959991|254.619995|258.179993|258.179993| 7048400|\n",
      "|2019-07-22|    258.75|262.149994|254.190002|255.679993|255.679993| 6842400|\n",
      "|2019-07-23|256.709991|260.480011|     254.5|260.170013|260.170013| 5023100|\n",
      "|2019-07-24|259.170013|266.070007|258.160004|264.880005|264.880005|11072800|\n",
      "|2019-07-25|     233.5|     234.5|225.550003|228.820007|228.820007|22418300|\n",
      "|2019-07-26|226.919998|230.259995|    222.25|228.039993|228.039993|10027700|\n",
      "|2019-07-29|227.089996|235.940002|226.029999|235.770004|235.770004| 9273300|\n",
      "|2019-07-30|232.899994|243.360001|232.179993|242.259995|242.259995| 8109000|\n",
      "|2019-07-31|     243.0|246.679993|236.649994|241.610001|241.610001| 9178200|\n",
      "|2019-08-01|242.649994|244.509995|231.770004|233.850006|233.850006| 8259500|\n",
      "|2019-08-02|231.350006|236.270004|229.229996|234.339996|234.339996| 6136500|\n",
      "|2019-08-05|229.600006|231.369995|225.779999|228.320007|228.320007| 7028300|\n",
      "|2019-08-06|231.880005|     232.5|    225.75|    230.75|    230.75| 5564200|\n",
      "|2019-08-07|     226.5|233.570007|225.800003|233.419998|233.419998| 4776500|\n",
      "|2019-08-08|234.449997|239.800003|232.649994|238.300003|238.300003| 5274300|\n",
      "|2019-08-09|236.050003|238.960007|233.809998|235.009995|235.009995| 3898200|\n",
      "+----------+----------+----------+----------+----------+----------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "tesla_data_file = 'TSLA.csv'\n",
    "\n",
    "tesla_df = sqlContext.read.load(tesla_data_file, \n",
    "                      format='com.databricks.spark.csv', \n",
    "                      header='true', \n",
    "                      inferSchema='true')\n",
    "\n",
    "tesla_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "|      Date|       Open|       High|        Low|      Close|   AdjClose| Volume|\n",
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "|2019-07-15|1146.859985|1150.819946|1139.400024|1150.339966|1150.339966| 903800|\n",
      "|2019-07-16|     1146.0|1158.579956|     1145.0|1153.579956|1153.579956|1238800|\n",
      "|2019-07-17|1150.969971|1158.359985| 1145.77002|1146.349976|1146.349976|1170000|\n",
      "|2019-07-18| 1141.73999| 1147.60498| 1132.72998|1146.329956|1146.329956|1291300|\n",
      "|2019-07-19|1148.189941|1151.140015|1129.619995|1130.099976|1130.099976|1647200|\n",
      "|2019-07-22|1133.449951|    1139.25| 1124.23999|1138.069946|1138.069946|1301500|\n",
      "|2019-07-23|     1144.0|1146.900024|1131.800049|1146.209961|1146.209961|1093700|\n",
      "|2019-07-24|1131.900024|     1144.0| 1126.98999|1137.810059|1137.810059|1589800|\n",
      "|2019-07-25|1137.819946|1141.699951|1120.920044|1132.119995|1132.119995|2209800|\n",
      "|2019-07-26|1224.040039|1265.550049|     1224.0|1250.410034|1250.410034|4805800|\n",
      "|2019-07-29|1241.050049|1247.369995| 1228.22998|1239.410034|1239.410034|2223700|\n",
      "|2019-07-30|1225.410034|1234.869995|1223.300049|1225.140015|1225.140015|1453300|\n",
      "|2019-07-31|     1223.0|     1234.0|1207.764038|1216.680054|1216.680054|1725500|\n",
      "|2019-08-01|1214.030029|1234.109985|1205.719971| 1209.01001| 1209.01001|1698500|\n",
      "|2019-08-02| 1200.73999|1206.900024|1188.939941| 1193.98999| 1193.98999|1645100|\n",
      "|2019-08-05|1170.040039| 1175.23999|1140.140015|1152.319946|1152.319946|2597500|\n",
      "|2019-08-06|1163.310059|1179.959961|     1160.0|1169.949951|1169.949951|1709400|\n",
      "|2019-08-07|     1156.0|1178.444946|1149.624023| 1173.98999| 1173.98999|1444300|\n",
      "|2019-08-08|1182.829956| 1205.01001| 1173.02002|1204.800049|1204.800049|1468000|\n",
      "|2019-08-09| 1197.98999|1203.880005|1183.603027| 1188.01001| 1188.01001|1065700|\n",
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "google_data_file = 'GOOG.csv'\n",
    "\n",
    "google_df = sqlContext.read.load(google_data_file, \n",
    "                      format='com.databricks.spark.csv', \n",
    "                      header='true', \n",
    "                      inferSchema='true')\n",
    "\n",
    "google_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "|      Date|       Open|       High|        Low|      Close|   AdjClose| Volume|\n",
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "|2019-07-15|2021.400024|2022.900024|2001.550049| 2020.98999| 2020.98999|2981300|\n",
      "|2019-07-16|2010.579956|2026.319946|2001.219971|2009.900024|2009.900024|2618200|\n",
      "|2019-07-17|2007.050049|     2012.0|1992.030029|1992.030029|1992.030029|2558800|\n",
      "|2019-07-18| 1980.01001|     1987.5|1951.550049|1977.900024|1977.900024|3504300|\n",
      "|2019-07-19|1991.209961|     1996.0| 1962.22998| 1964.52002| 1964.52002|3185600|\n",
      "|2019-07-22|1971.140015|     1989.0| 1958.26001|1985.630005|1985.630005|2900000|\n",
      "|2019-07-23| 1995.98999|1997.790039|1973.130005| 1994.48999| 1994.48999|2703500|\n",
      "|2019-07-24|1969.300049|2001.300049|1965.869995|2000.810059|2000.810059|2631300|\n",
      "|2019-07-25|     2001.0|2001.199951|1972.719971|1973.819946|1973.819946|4136500|\n",
      "|2019-07-26|     1942.0|1950.900024| 1924.51001|1943.050049|1943.050049|4927100|\n",
      "|2019-07-29|     1930.0| 1932.22998|1890.540039|1912.449951|1912.449951|4493200|\n",
      "|2019-07-30|1891.119995|1909.890015| 1883.47998|1898.530029|1898.530029|2910900|\n",
      "|2019-07-31|1898.109985|1899.550049|1849.439941|1866.780029|1866.780029|4470700|\n",
      "|2019-08-01|1871.719971|1897.920044| 1844.01001|1855.319946|1855.319946|4713300|\n",
      "|2019-08-02|1845.069946|1846.359985| 1808.02002| 1823.23999| 1823.23999|4956200|\n",
      "|2019-08-05|1770.219971|1788.670044|1748.780029|1765.130005|1765.130005|6058200|\n",
      "|2019-08-06| 1792.22998| 1793.77002|1753.400024|1787.829956|1787.829956|5070300|\n",
      "|2019-08-07| 1773.98999|1798.930054|     1757.0|1793.400024|1793.400024|4526900|\n",
      "|2019-08-08|     1806.0| 1834.26001|1798.109985|1832.890015|1832.890015|3701200|\n",
      "|2019-08-09|1828.949951|1831.089966|1802.219971|1807.579956|1807.579956|2879800|\n",
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "amazon_data_file = 'AMZN.csv'\n",
    "\n",
    "amazon_df = sqlContext.read.load(amazon_data_file, \n",
    "                      format='com.databricks.spark.csv', \n",
    "                      header='true', \n",
    "                      inferSchema='true')\n",
    "\n",
    "amazon_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------------------+\n",
      "|year|     avg(AdjClose)|\n",
      "+----+------------------+\n",
      "|2019|1800.6161329411755|\n",
      "|2020| 2236.414478798507|\n",
      "+----+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "import datetime\n",
    "from pyspark.sql.functions import year, month, dayofmonth\n",
    "\n",
    "# Average closing price per year for AMZN\n",
    "# stocksDF.select(year($\"dt\").alias(\"yr\"), $\"adjcloseprice\").groupBy(\"yr\").avg(\"adjcloseprice\").orderBy(desc(\"yr\")).show\n",
    "\n",
    "amazon_df.select(year(\"Date\").alias(\"year\"), \"AdjClose\").groupby(\"year\").avg(\"AdjClose\").sort(\"year\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+------------------+\n",
      "|year|month|     avg(AdjClose)|\n",
      "+----+-----+------------------+\n",
      "|2019|    7|1964.6846265384618|\n",
      "|2019|    8|1793.6027220909093|\n",
      "|2019|    9|     1799.12099615|\n",
      "|2019|   10|1752.3317498695653|\n",
      "|2019|   11|      1774.2939941|\n",
      "|2019|   12|1785.7728446190476|\n",
      "|2020|    1|1884.2376128571425|\n",
      "|2020|    2|2066.1752672631574|\n",
      "|2020|    3|1872.3104358636365|\n",
      "|2020|    4|2228.7052408571426|\n",
      "|2020|    5|2394.1840209499996|\n",
      "|2020|    6|      2613.5454545|\n",
      "|2020|    7| 3053.100016222222|\n",
      "+----+-----+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Compute the average closing price per month for apc\n",
    "# stocksDF.select(year($\"dt\").alias(\"yr\"),month($\"dt\").alias(\"mo\"), $\"adjcloseprice\")\n",
    "# .groupBy(\"yr\",\"mo\").agg(avg(\"adjcloseprice\")).orderBy(desc(\"yr\"),desc(\"mo\")).show\n",
    "\n",
    "amazon_df.select(year(\"Date\").alias(\"year\"),\n",
    "                month(\"Date\").alias(\"month\"),\n",
    "                \"AdjClose\").groupby(\"year\", \"month\").avg(\"AdjClose\").sort(\"year\", \"month\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "# register the DataFrames as  temp views\n",
    "\n",
    "amazon_df.registerTempTable(\"amazon_stocks\")\n",
    "google_df.registerTempTable(\"google_stocks\")\n",
    "tesla_df.registerTempTable(\"tesla_stocks\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "|      Date|       Open|       High|        Low|      Close|   AdjClose| Volume|\n",
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "|2019-07-15|2021.400024|2022.900024|2001.550049| 2020.98999| 2020.98999|2981300|\n",
      "|2019-07-16|2010.579956|2026.319946|2001.219971|2009.900024|2009.900024|2618200|\n",
      "|2019-07-17|2007.050049|     2012.0|1992.030029|1992.030029|1992.030029|2558800|\n",
      "|2019-07-18| 1980.01001|     1987.5|1951.550049|1977.900024|1977.900024|3504300|\n",
      "|2019-07-19|1991.209961|     1996.0| 1962.22998| 1964.52002| 1964.52002|3185600|\n",
      "+----------+-----------+-----------+-----------+-----------+-----------+-------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT * FROM amazon_stocks\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+---+------------------+\n",
      "|  yr| mo|     avg(AdjClose)|\n",
      "+----+---+------------------+\n",
      "|2019| 10|1752.3317498695653|\n",
      "|2020|  6|      2613.5454545|\n",
      "|2020|  3|1872.3104358636365|\n",
      "|2019|  8|1793.6027220909093|\n",
      "|2020|  4|2228.7052408571426|\n",
      "|2020|  1|1884.2376128571425|\n",
      "|2019|  9|     1799.12099615|\n",
      "|2019| 12|1785.7728446190476|\n",
      "|2020|  7| 3053.100016222222|\n",
      "|2020|  2|2066.1752672631574|\n",
      "|2019|  7|1964.6846265384618|\n",
      "|2019| 11|      1774.2939941|\n",
      "|2020|  5|2394.1840209499996|\n",
      "+----+---+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Calculate and display the average closing price per month for XOM ordered by year,month \n",
    "\n",
    "sqlContext.sql(\"\"\"SELECT year(amazon_stocks.Date) as yr, month(amazon_stocks.Date) as mo, avg(amazon_stocks.AdjClose) from amazon_stocks group By year(amazon_stocks.Date), month(amazon_stocks.Date)\"\"\").show()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------+-----------+------------------+\n",
      "|      Date|       Open|      Close|            spydif|\n",
      "+----------+-----------+-----------+------------------+\n",
      "|2019-07-16|     1146.0|1153.579956| 7.579956000000038|\n",
      "|2019-07-17|1150.969971|1146.349976| 4.619995000000017|\n",
      "|2019-07-18| 1141.73999|1146.329956| 4.589966000000004|\n",
      "|2019-07-19|1148.189941|1130.099976| 18.08996500000012|\n",
      "|2019-07-22|1133.449951|1138.069946| 4.619995000000017|\n",
      "|2019-07-24|1131.900024|1137.810059|  5.91003499999988|\n",
      "|2019-07-25|1137.819946|1132.119995|5.6999510000000555|\n",
      "|2019-07-26|1224.040039|1250.410034|26.369995000000017|\n",
      "|2019-07-31|     1223.0|1216.680054| 6.319946000000073|\n",
      "|2019-08-01|1214.030029| 1209.01001|5.0200190000000475|\n",
      "|2019-08-02| 1200.73999| 1193.98999|              6.75|\n",
      "|2019-08-05|1170.040039|1152.319946|17.720092999999906|\n",
      "|2019-08-06|1163.310059|1169.949951| 6.639892000000145|\n",
      "|2019-08-07|     1156.0| 1173.98999|17.989990000000034|\n",
      "|2019-08-08|1182.829956|1204.800049|21.970092999999906|\n",
      "|2019-08-09| 1197.98999| 1188.01001| 9.979980000000069|\n",
      "|2019-08-12|1179.209961|1174.709961|               4.5|\n",
      "|2019-08-13|1171.459961| 1197.27002| 25.81005899999991|\n",
      "|2019-08-14|1176.310059|1164.290039|12.020019999999931|\n",
      "|2019-08-19|1190.089966|1198.449951| 8.359985000000052|\n",
      "+----------+-----------+-----------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# When did the closing price for SPY go up or down by more than 2 dollars?\n",
    "\n",
    "sqlContext.sql(\"SELECT google_stocks.Date, google_stocks.Open, google_stocks.Close, abs(google_stocks.Close - google_stocks.Open) as spydif FROM google_stocks WHERE abs(google_stocks.Close - google_stocks.Open) > 4 \").show()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------+-------------+\n",
      "|  yr|max(AdjClose)|min(AdjClose)|\n",
      "+----+-------------+-------------+\n",
      "|2019|   430.940002|   211.399994|\n",
      "|2020|  1544.650024|   361.220001|\n",
      "+----+-------------+-------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#What was the max, min closing price for SPY and XOM by Year?</font> \n",
    "\n",
    "sqlContext.sql(\"SELECT year(tesla_stocks.Date) as yr, max(tesla_stocks.AdjClose), min(tesla_stocks.AdjClose) FROM tesla_stocks group By year(tesla_stocks.Date)\").show()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(2) HashAggregate(keys=[year(cast(Date#133 as date))#524], functions=[max(AdjClose#138), min(AdjClose#138)])\n",
      "+- Exchange hashpartitioning(year(cast(Date#133 as date))#524, 200), true, [id=#324]\n",
      "   +- *(1) HashAggregate(keys=[year(cast(Date#133 as date)) AS year(cast(Date#133 as date))#524], functions=[partial_max(AdjClose#138), partial_min(AdjClose#138)])\n",
      "      +- FileScan csv [Date#133,AdjClose#138] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex[file:/home/jovyan/work/TSLA.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<Date:string,AdjClose:double>\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT year(tesla_stocks.Date) as yr, max(tesla_stocks.AdjClose), min(tesla_stocks.AdjClose) FROM tesla_stocks group By year(tesla_stocks.Date)\").explain()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+-----------+-----------+\n",
      "|      Date|teslaclose|amazonclose|googleclose|\n",
      "+----------+----------+-----------+-----------+\n",
      "|2019-07-15|     253.5| 2020.98999|1150.339966|\n",
      "|2019-07-16|252.380005|2009.900024|1153.579956|\n",
      "|2019-07-17|254.860001|1992.030029|1146.349976|\n",
      "|2019-07-18|253.539993|1977.900024|1146.329956|\n",
      "|2019-07-19|258.179993| 1964.52002|1130.099976|\n",
      "|2019-07-22|255.679993|1985.630005|1138.069946|\n",
      "|2019-07-23|260.170013| 1994.48999|1146.209961|\n",
      "|2019-07-24|264.880005|2000.810059|1137.810059|\n",
      "|2019-07-25|228.820007|1973.819946|1132.119995|\n",
      "|2019-07-26|228.039993|1943.050049|1250.410034|\n",
      "|2019-07-29|235.770004|1912.449951|1239.410034|\n",
      "|2019-07-30|242.259995|1898.530029|1225.140015|\n",
      "|2019-07-31|241.610001|1866.780029|1216.680054|\n",
      "|2019-08-01|233.850006|1855.319946| 1209.01001|\n",
      "|2019-08-02|234.339996| 1823.23999| 1193.98999|\n",
      "|2019-08-05|228.320007|1765.130005|1152.319946|\n",
      "|2019-08-06|    230.75|1787.829956|1169.949951|\n",
      "|2019-08-07|233.419998|1793.400024| 1173.98999|\n",
      "|2019-08-08|238.300003|1832.890015|1204.800049|\n",
      "|2019-08-09|235.009995|1807.579956| 1188.01001|\n",
      "+----------+----------+-----------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Join all stock closing prices in order to compare\n",
    "\n",
    "joinclose=sqlContext.sql(\"SELECT tesla_stocks.Date, tesla_stocks.AdjClose as teslaclose, amazon_stocks.AdjClose as amazonclose, google_stocks.AdjClose as googleclose from tesla_stocks join google_stocks on tesla_stocks.Date = google_stocks.Date join amazon_stocks on tesla_stocks.Date = amazon_stocks.Date\").cache()\n",
    "\n",
    "joinclose.show()\n",
    "\n",
    "joinclose.registerTempTable(\"joinclose\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+------------------+------------------+------------------+\n",
      "|  yr|        teslaclose|       amazonclose|       googleclose|\n",
      "+----+------------------+------------------+------------------+\n",
      "|2019|283.62126070588243|1800.6161329411755|1245.3833654621849|\n",
      "|2020| 761.8206739179104| 2236.414478798507|1362.8286906865671|\n",
      "+----+------------------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlContext.sql(\"SELECT year(joinclose.Date) as yr, avg(joinclose.teslaclose) as teslaclose, avg(joinclose.amazonclose) as amazonclose, avg(joinclose.googleclose) as googleclose from joinclose group By year(joinclose.Date) order by year(joinclose.Date)\").show()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "joinclose.write.format(\"parquet\").save(\"joinstocks.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+-----------+-----------+\n",
      "|      Date|teslaclose|amazonclose|googleclose|\n",
      "+----------+----------+-----------+-----------+\n",
      "|2019-07-15|     253.5| 2020.98999|1150.339966|\n",
      "|2019-07-16|252.380005|2009.900024|1153.579956|\n",
      "|2019-07-17|254.860001|1992.030029|1146.349976|\n",
      "|2019-07-18|253.539993|1977.900024|1146.329956|\n",
      "|2019-07-19|258.179993| 1964.52002|1130.099976|\n",
      "|2019-07-22|255.679993|1985.630005|1138.069946|\n",
      "|2019-07-23|260.170013| 1994.48999|1146.209961|\n",
      "|2019-07-24|264.880005|2000.810059|1137.810059|\n",
      "|2019-07-25|228.820007|1973.819946|1132.119995|\n",
      "|2019-07-26|228.039993|1943.050049|1250.410034|\n",
      "|2019-07-29|235.770004|1912.449951|1239.410034|\n",
      "|2019-07-30|242.259995|1898.530029|1225.140015|\n",
      "|2019-07-31|241.610001|1866.780029|1216.680054|\n",
      "|2019-08-01|233.850006|1855.319946| 1209.01001|\n",
      "|2019-08-02|234.339996| 1823.23999| 1193.98999|\n",
      "|2019-08-05|228.320007|1765.130005|1152.319946|\n",
      "|2019-08-06|    230.75|1787.829956|1169.949951|\n",
      "|2019-08-07|233.419998|1793.400024| 1173.98999|\n",
      "|2019-08-08|238.300003|1832.890015|1204.800049|\n",
      "|2019-08-09|235.009995|1807.579956| 1188.01001|\n",
      "+----------+----------+-----------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "final_df = sqlContext.read.parquet(\"joinstocks.parquet\")\n",
    "final_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Date: string (nullable = true)\n",
      " |-- teslaclose: double (nullable = true)\n",
      " |-- amazonclose: double (nullable = true)\n",
      " |-- googleclose: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "final_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-----+------------------+------------------+------------------+\n",
      "|year|month|   avg(teslaclose)|  avg(amazonclose)|  avg(googleclose)|\n",
      "+----+-----+------------------+------------------+------------------+\n",
      "|2019|    7|248.43769253846148|1964.6846265384618|1170.1961483076923|\n",
      "|2019|    8|225.10272704545451|1793.6027220909093|1180.6868120454546|\n",
      "|2019|    9|237.26149830000003|     1799.12099615|     1220.83952035|\n",
      "|2019|   10| 266.3547840434783|1752.3317498695653|1232.7117442608696|\n",
      "|2019|   11|338.30000000000007|      1774.2939941|     1304.27899165|\n",
      "|2019|   12| 377.6947631904762|1785.7728446190476|1340.8676351904762|\n",
      "|2020|    1| 528.6590503809524|1884.2376128571425|1436.6537968571424|\n",
      "|2020|    2| 797.4468415263159|2066.1752672631574|1464.1105184736841|\n",
      "|2020|    3| 559.1013613181818|1872.3104358636365|1188.3940984545457|\n",
      "|2020|    4| 663.5985761428572|2228.7052408571426|1234.1404797142854|\n",
      "|2020|    5|      799.42549745|2394.1840209499996|1381.1137511999998|\n",
      "|2020|    6| 963.5422779545456|      2613.5454545|1431.0477184545452|\n",
      "|2020|    7|1378.7111273333333| 3053.100016222222|1496.0299885555555|\n",
      "+----+-----+------------------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Average of each by month\n",
    "final_df.select(year(\"Date\").alias(\"year\"), \n",
    "                 month(\"Date\").alias(\"month\"), \n",
    "                 \"teslaclose\", \"amazonclose\", \n",
    "                 \"googleclose\").groupby(\"year\", \n",
    "                                        \"month\").avg(\"teslaclose\", \n",
    "                                                     \"amazonclose\", \n",
    "                                                     \"googleclose\").sort(\"year\", \n",
    "                                                                         \"month\").show()\n",
    "                "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "== Physical Plan ==\n",
      "*(3) Sort [year#886 ASC NULLS FIRST, month#887 ASC NULLS FIRST], true, 0\n",
      "+- Exchange rangepartitioning(year#886 ASC NULLS FIRST, month#887 ASC NULLS FIRST, 200), true, [id=#543]\n",
      "   +- *(2) HashAggregate(keys=[year#886, month#887], functions=[avg(teslaclose#806), avg(amazonclose#807), avg(googleclose#808)])\n",
      "      +- Exchange hashpartitioning(year#886, month#887, 200), true, [id=#539]\n",
      "         +- *(1) HashAggregate(keys=[year#886, month#887], functions=[partial_avg(teslaclose#806), partial_avg(amazonclose#807), partial_avg(googleclose#808)])\n",
      "            +- *(1) Project [year(cast(Date#805 as date)) AS year#886, month(cast(Date#805 as date)) AS month#887, teslaclose#806, amazonclose#807, googleclose#808]\n",
      "               +- *(1) ColumnarToRow\n",
      "                  +- FileScan parquet [Date#805,teslaclose#806,amazonclose#807,googleclose#808] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/home/jovyan/work/joinstocks.parquet], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<Date:string,teslaclose:double,amazonclose:double,googleclose:double>\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "final_df.select(year(\"Date\").alias(\"year\"), \n",
    "                 month(\"Date\").alias(\"month\"), \n",
    "                 \"teslaclose\", \"amazonclose\", \n",
    "                 \"googleclose\").groupby(\"year\", \n",
    "                                        \"month\").avg(\"teslaclose\", \n",
    "                                                     \"amazonclose\", \n",
    "                                                     \"googleclose\").sort(\"year\", \n",
    "                                                                         \"month\").explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "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.8.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}