averages_of_aggregated_counts.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "uxYRVWTNBvj6"
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.dataframe import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "NqMXDwR1Bvj-"
   },
   "outputs": [],
   "source": [
    "spark = SparkSession.builder \\\n",
    "    .appName(\"taxi\") \\\n",
    "    .getOrCreate() "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "eVrJK6lbBvkB"
   },
   "outputs": [],
   "source": [
    "ride_df = spark.read.csv('data.csv', inferSchema=True, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "6V4Us0_aBvkD",
    "outputId": "69fa9ee0-0cae-4465-f1ff-1682cca132e6"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ts: timestamp (nullable = true)\n",
      " |-- number: string (nullable = true)\n",
      " |-- pick_lat: double (nullable = true)\n",
      " |-- pick_lng: double (nullable = true)\n",
      " |-- drop_lat: double (nullable = true)\n",
      " |-- drop_lng: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ride_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "muJGpkHhBvkK",
    "outputId": "6b0f8ce7-4420-48a5-b826-027402578762"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8381556"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ride_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "FEuXqmzoBvkM"
   },
   "outputs": [],
   "source": [
    "ride_df.createOrReplaceTempView(\"rider\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "AXmWIPPlBvkP",
    "outputId": "bdff88bd-5e4d-45e0-9532-a62784d0d4dc"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Number of taxis 94274\n"
     ]
    }
   ],
   "source": [
    "print(\"Number of taxis\", spark.sql('SELECT distinct number from rider where number >=0').count())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "IBvxcWh2BvkU",
    "outputId": "1078fbcc-8a7b-4341-ebf2-c989ae831314"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ts: timestamp (nullable = true)\n",
      " |-- number: integer (nullable = true)\n",
      " |-- pick_lat: double (nullable = true)\n",
      " |-- pick_lng: double (nullable = true)\n",
      " |-- drop_lat: double (nullable = true)\n",
      " |-- drop_lng: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ride_df = ride_df.withColumn(\"number\", ride_df[\"number\"].cast(IntegerType()))\n",
    "ride_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "djVU8_gWBvkW"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8381183"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# cleaned the dataset\n",
    "ride_df = spark.sql('SELECT * from rider where number >=0')\n",
    "ride_df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "ride_df.createOrReplaceTempView('rider')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "hVxHIdI9BvkZ",
    "outputId": "2efeeb21-def7-415b-c71e-c9e10e493cd5"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+------------------+------------------+------------------+------------------+\n",
      "|summary|            number|          pick_lat|          pick_lng|          drop_lat|          drop_lng|\n",
      "+-------+------------------+------------------+------------------+------------------+------------------+\n",
      "|  count|           8381183|           8381183|           8381183|           8381183|           8381183|\n",
      "|   mean| 49836.32284571283|13.010247505849685| 77.62874600026429|13.010846611327048| 77.60708484316284|\n",
      "| stddev|28903.490367448314|0.9032978033556502|0.4490997661942238|0.9516671807115192|1.4753016402249166|\n",
      "|    min|             00000|         -48.77217|         -93.95859|        -57.946835|        -127.81789|\n",
      "|    max|             99999|          67.69665|          92.81412|          61.52401|         174.88597|\n",
      "+-------+------------------+------------------+------------------+------------------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ride_df.describe().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "WlV2MceQBvkb",
    "outputId": "1fba584e-e4dc-4535-f069-2e4f8662eaac"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----------+\n",
      "|           per_hour|avg(rides)|\n",
      "+-------------------+----------+\n",
      "|2018-04-07 01:00:00|      20.0|\n",
      "|2018-04-07 02:00:00|       5.0|\n",
      "|2018-04-07 03:00:00|       1.0|\n",
      "|2018-04-07 05:00:00|       6.0|\n",
      "|2018-04-07 06:00:00|      66.0|\n",
      "|2018-04-07 07:00:00|     178.0|\n",
      "|2018-04-07 08:00:00|     319.0|\n",
      "|2018-04-07 09:00:00|     698.0|\n",
      "|2018-04-07 10:00:00|     506.0|\n",
      "|2018-04-07 11:00:00|     611.0|\n",
      "|2018-04-07 12:00:00|     677.0|\n",
      "|2018-04-07 13:00:00|     551.0|\n",
      "|2018-04-07 14:00:00|     549.0|\n",
      "|2018-04-07 15:00:00|     472.0|\n",
      "|2018-04-07 16:00:00|     544.0|\n",
      "|2018-04-07 17:00:00|     607.0|\n",
      "|2018-04-07 18:00:00|     696.0|\n",
      "|2018-04-07 19:00:00|     945.0|\n",
      "|2018-04-07 20:00:00|     936.0|\n",
      "|2018-04-07 21:00:00|     412.0|\n",
      "+-------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# averages of aggregated counts per hour\n",
    "spark.sql(\"select per_hour, avg(rides) from(select date_trunc('hour',ts) per_hour, count(number) rides from rider group by 1)group by 1 order by 1\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "h7wdEkgOBvkd",
    "outputId": "a450ef0a-ac55-4be6-82fa-6c6e94341591"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----------+\n",
      "|            per_day|avg(rides)|\n",
      "+-------------------+----------+\n",
      "|2018-04-07 00:00:00|    9374.0|\n",
      "|2018-04-08 00:00:00|    5716.0|\n",
      "|2018-04-09 00:00:00|   11552.0|\n",
      "|2018-04-10 00:00:00|   11186.0|\n",
      "|2018-04-11 00:00:00|   12126.0|\n",
      "|2018-04-12 00:00:00|   10359.0|\n",
      "|2018-04-13 00:00:00|   13468.0|\n",
      "|2018-04-14 00:00:00|    6588.0|\n",
      "|2018-04-15 00:00:00|    4192.0|\n",
      "|2018-04-16 00:00:00|   10534.0|\n",
      "|2018-04-17 00:00:00|    9454.0|\n",
      "|2018-04-18 00:00:00|    9644.0|\n",
      "|2018-04-19 00:00:00|   12261.0|\n",
      "|2018-04-20 00:00:00|   10061.0|\n",
      "|2018-04-21 00:00:00|    6916.0|\n",
      "|2018-04-22 00:00:00|    3225.0|\n",
      "|2018-04-23 00:00:00|    9049.0|\n",
      "|2018-04-24 00:00:00|    8205.0|\n",
      "|2018-04-25 00:00:00|    9159.0|\n",
      "|2018-04-26 00:00:00|    9009.0|\n",
      "+-------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# averages of aggregated counts per day\n",
    "spark.sql(\"select per_day, avg(rides) from(select date_trunc('day',ts) per_day, count(number) rides from rider group by 1)group by 1 order by 1\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "UOqkiV_1Bvkf",
    "outputId": "a169bf34-c7ed-4924-a9d6-b723d7b2de66"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----------+\n",
      "|           per_week|avg(rides)|\n",
      "+-------------------+----------+\n",
      "|2018-04-02 00:00:00|   15090.0|\n",
      "|2018-04-09 00:00:00|   69471.0|\n",
      "|2018-04-16 00:00:00|   62095.0|\n",
      "|2018-04-23 00:00:00|   55833.0|\n",
      "|2018-04-30 00:00:00|   47553.0|\n",
      "|2018-05-07 00:00:00|   51371.0|\n",
      "|2018-05-14 00:00:00|   56614.0|\n",
      "|2018-05-21 00:00:00|   57283.0|\n",
      "|2018-05-28 00:00:00|   63352.0|\n",
      "|2018-06-04 00:00:00|   59812.0|\n",
      "|2018-06-11 00:00:00|   60669.0|\n",
      "|2018-06-18 00:00:00|   63241.0|\n",
      "|2018-06-25 00:00:00|   60759.0|\n",
      "|2018-07-02 00:00:00|   61134.0|\n",
      "|2018-07-09 00:00:00|   64492.0|\n",
      "|2018-07-16 00:00:00|   65746.0|\n",
      "|2018-07-23 00:00:00|   67281.0|\n",
      "|2018-07-30 00:00:00|   64243.0|\n",
      "|2018-08-06 00:00:00|   74065.0|\n",
      "|2018-08-13 00:00:00|   79563.0|\n",
      "+-------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# averages of aggregated counts per week\n",
    "spark.sql(\"select per_week, avg(rides) from(select date_trunc('week',ts) per_week, count(number) rides from rider group by 1)group by 1 order by 1\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "8YeFUcKeBvki",
    "outputId": "76de1d92-96eb-4876-9c6d-32fd10d36700"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+----------+\n",
      "|          per_month|avg(rides)|\n",
      "+-------------------+----------+\n",
      "|2018-04-01 00:00:00|  209246.0|\n",
      "|2018-05-01 00:00:00|  246221.0|\n",
      "|2018-06-01 00:00:00|  263026.0|\n",
      "|2018-07-01 00:00:00|  282412.0|\n",
      "|2018-08-01 00:00:00|  389824.0|\n",
      "|2018-09-01 00:00:00|  573085.0|\n",
      "|2018-10-01 00:00:00|  623563.0|\n",
      "|2018-11-01 00:00:00|  610488.0|\n",
      "|2018-12-01 00:00:00|  671155.0|\n",
      "|2019-01-01 00:00:00|  876294.0|\n",
      "|2019-02-01 00:00:00| 1293469.0|\n",
      "|2019-03-01 00:00:00| 1761307.0|\n",
      "|2019-04-01 00:00:00|  581093.0|\n",
      "+-------------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# averages of aggregated counts per month\n",
    "spark.sql(\"select per_month, avg(rides) from(select date_trunc('month',ts) per_month, count(number) rides from rider group by 1)group by 1 order by 1\").show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "oOijVne0Bvkk"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-------------------+------+----------+-----------------+---------+-----------------+\n",
      "|                 ts|            ts_week|number|  pick_lat|         pick_lng| drop_lat|         drop_lng|\n",
      "+-------------------+-------------------+------+----------+-----------------+---------+-----------------+\n",
      "|2018-04-07 07:07:17|2018-04-02 00:00:00| 14626|12.3136215|76.65819499999998|12.287301|76.60228000000002|\n",
      "|2018-04-07 07:32:27|2018-04-02 00:00:00| 85490| 12.943947|        77.560745|12.954014|         77.54377|\n",
      "|2018-04-07 07:36:44|2018-04-02 00:00:00| 05408| 12.899603|          77.5873| 12.93478|         77.56995|\n",
      "|2018-04-07 07:38:00|2018-04-02 00:00:00| 58940| 12.918229|77.60754399999998|12.968971|        77.636375|\n",
      "|2018-04-07 07:39:29|2018-04-02 00:00:00| 05408|  12.89949|77.58726999999998| 12.93478|         77.56995|\n",
      "|2018-04-07 07:43:08|2018-04-02 00:00:00| 05408| 12.899421|        77.587326| 12.93478|         77.56995|\n",
      "|2018-04-07 07:43:55|2018-04-02 00:00:00| 50266| 12.898679|77.60434000000002|12.877949|          77.5959|\n",
      "|2018-04-07 07:52:31|2018-04-02 00:00:00| 58940| 12.918229|77.60754399999998|12.968971|        77.636375|\n",
      "|2018-04-07 07:52:42|2018-04-02 00:00:00| 58940| 12.918229|77.60754399999998|12.968971|        77.636375|\n",
      "|2018-04-07 07:53:23|2018-04-02 00:00:00| 28126|  12.91184|         77.60225|12.940866|         77.54071|\n",
      "|2018-04-07 07:55:05|2018-04-02 00:00:00| 99251|  12.87466|77.61950999999998|12.896871|         77.60847|\n",
      "|2018-04-07 07:55:24|2018-04-02 00:00:00| 99251|  12.87466|77.61950999999998|12.896871|         77.60847|\n",
      "|2018-04-07 08:00:04|2018-04-02 00:00:00| 34808| 12.989711|         77.65381|12.939158|         77.73467|\n",
      "|2018-04-07 08:00:16|2018-04-02 00:00:00| 34808| 12.989711|         77.65381|12.939158|         77.73467|\n",
      "|2018-04-07 08:03:16|2018-04-02 00:00:00| 89714| 12.868537|         77.65304|12.972006|         77.59487|\n",
      "|2018-04-07 08:03:24|2018-04-02 00:00:00| 89714| 12.868537|         77.65304|12.972006|         77.59487|\n",
      "|2018-04-07 08:07:16|2018-04-02 00:00:00| 82060| 12.987069|77.57703000000002|12.970017|        77.577934|\n",
      "|2018-04-07 08:08:57|2018-04-02 00:00:00| 18815| 12.933479|         77.57087|12.961353|         77.57457|\n",
      "|2018-04-07 08:11:35|2018-04-02 00:00:00| 38288| 12.886039|         77.64894|12.902692|         77.62253|\n",
      "|2018-04-07 08:17:24|2018-04-02 00:00:00| 80401| 12.990636|         77.67494|12.962719|          77.5876|\n",
      "+-------------------+-------------------+------+----------+-----------------+---------+-----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "export_week = spark.sql(\"select ts,  date_trunc('week', ts) ts_week, number, pick_lat, pick_lng, drop_lat, drop_lng from rider\")\n",
    "export_week.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "O-vKADGxBvkm"
   },
   "outputs": [],
   "source": [
    "# dumping results for cohort analysis\n",
    "export_week.coalesce(1).write.csv('result-dumps-week')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "colab": {},
    "colab_type": "code",
    "id": "bzlbe8UgBvkp",
    "outputId": "00606bbd-0cb9-48d0-ee96-292a7a8ca727"
   },
   "outputs": [],
   "source": [
    "# trying out window funcs\n",
    "spark.sql(\"select count(number), date_trunc('week',first_time) from(select distinct number, first_value(ts) over (partition by number order by ts) as first_time from rider) group by 2 order by 2\").show(53)"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "name": "averages_of_aggregated_counts.ipynb",
   "provenance": []
  },
  "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}