NYCTaxi_ak_DataFrames.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exploring the New York City Taxi Data with Arkouda + Pandas/NumPy\n",
    "\n",
    "This notebook shows some examples of how to interoperate between Pandas and Arkouda at a small scale on a few-GB workstation. This same notebook would run with a multi-node Arkouda instance on an HPC with TB of data.\n",
    "\n",
    "Arkouda is not trying to replace Pandas but to allow for some Pandas-style operation at a much larger scale. In our experience Pandas can handle dataframes up to about **500 million rows** on a sufficently capable compute server before performance becomes a real issue. Arkouda breaks the shared memory paradigm and scales its operations to distributed dataframes with **hundreds of billions of rows**, maybe even a trillion. In practice we have run Arkouda server operations on columns of one trillion elements running on 512 compute nodes. This yielded a **>20TB dataframe** in Arkouda.\n",
    "\n",
    "**Outline**\n",
    "- Data Preparation\n",
    "  - Get Data\n",
    "  - Convert Data\n",
    "  - Load Data\n",
    "- Data Exploration\n",
    "  - Summarization\n",
    "  - Histograms\n",
    "  - Logical Indexing/Filtering\n",
    "  - Time Data\n",
    "  - Lookup Tables\n",
    "  - GroupBy-Aggregate\n",
    "  - Broadcast\n",
    "  - Integrate with Pandas\n",
    "\n",
    "# Data Preparation\n",
    "\n",
    "## Download New York City Taxi Data\n",
    "----------------------------------\n",
    "[Yellow Trips Data Dictionary](https://www1.nyc.gov/assets/tlc/downloads/pdf/data_dictionary_trip_records_yellow.pdf)\n",
    "\n",
    "[NYC Yellow Taxi Trip Records Jan 2020](https://s3.amazonaws.com/nyc-tlc/trip+data/yellow_tripdata_2020-01.csv)\n",
    "\n",
    "[Green Trips Data Dictionary](https://www1.nyc.gov/assets/tlc/downloads/pdf/data_dictionary_trip_records_green.pdf)\n",
    "\n",
    "[NYC Green  Taxi Trip Records Jan 2020](https://s3.amazonaws.com/nyc-tlc/trip+data/green_tripdata_2020-01.csv)\n",
    "\n",
    "[NYC Taxi Zone Lookup Table](https://s3.amazonaws.com/nyc-tlc/misc/taxi+_zone_lookup.csv)\n",
    "\n",
    "[NYC Taxi Zone Shapefile](https://s3.amazonaws.com/nyc-tlc/misc/taxi_zones.zip)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Convert and Load Data\n",
    "\n",
    "Currently, `ak.DataFrame` does not have a `from_pandas()` method. Because of this, we load from the csv loop over the columns building a dictionary that is used to build an Arkouda DataFrame. In the future, these methods will be available.\n",
    "\n",
    "#### Additional Dtypes in Arkouda\n",
    "Can cast/convert to these after loading raw data\n",
    "* bool\n",
    "* Datetime (from int64)\n",
    "* Timedelta (from int64)\n",
    "\n",
    "#### Prefer Integers!\n",
    "They are fast and versatile (usable with GroupBy, Datetime, Timedelta, bit ops, etc.)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Describe Data Format"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!head /Users/ethandebandi/Documents/test_data/green_tripdata_2020-01.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%file NYCTaxi_format.py\n",
    "\n",
    "import numpy as np\n",
    "\n",
    "OPTIONS = {}\n",
    "\n",
    "def YNint(yn):\n",
    "    return (0, 1)[yn.upper() in 'YES']\n",
    "\n",
    "def nullint(x):\n",
    "    try:\n",
    "        return np.int64(x)\n",
    "    except:\n",
    "        return np.int64(-1)\n",
    "\n",
    "yellow_format = {'sep': ',',\n",
    "                 'header': 0,\n",
    "                 'parse_dates':['tpep_dropoff_datetime', 'tpep_pickup_datetime'],\n",
    "                 'infer_datetime_format': True,\n",
    "                 'converters': {'store_and_fwd_flag': YNint,\n",
    "                                'VendorID': nullint,\n",
    "                                'RatecodeID': nullint,\n",
    "                                'PULocationID': nullint,\n",
    "                                'DOLocationID': nullint,\n",
    "                                'passenger_count': nullint,\n",
    "                                'payment_type': nullint,\n",
    "                                'trip_type': nullint}}\n",
    "\n",
    "OPTIONS['yellow'] = yellow_format\n",
    "\n",
    "green_format = yellow_format.copy()\n",
    "green_format['parse_dates'] = ['lpep_dropoff_datetime', 'lpep_pickup_datetime']\n",
    "OPTIONS['green'] = green_format"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### CSV --> Pandas --> Arkouda\n",
    "Pandas has a very good CSV reader, so we will use that.\n",
    "\n",
    "It might be worth noting that we should add an arkouda wrapper function to do what will be done here, ak.from_pandas() & ak.from_csv()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import NYCTaxi_format as taxi\n",
    "import arkouda as ak\n",
    "ak.connect(connect_url=\"tcp://localhost:5555\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pdgreen = pd.read_csv('/Users/ethandebandi/Documents/test_data/green_tripdata_2020-01.csv', **taxi.OPTIONS['green'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def ak_from_pandas(df):\n",
    "    #first we create a dictionary mapping column names to column data\n",
    "    ak_dict = {}\n",
    "    for cname in df.keys():\n",
    "        if df[cname].dtype.name == 'object':\n",
    "            ak_dict[cname] = ak.from_series(df[cname],dtype=np.str)\n",
    "        else:\n",
    "            ak_dict[cname] = ak.from_series(df[cname])\n",
    "    \n",
    "    return ak.DataFrame(ak_dict)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Create an arkouda DataFrame\n",
    "ak_green = ak_from_pandas(pdgreen)\n",
    "ak_green"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exploreation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Descriptive Statistics"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def describe(x):\n",
    "    fmt = 'mean: {}\\nstd : {}\\nmin : {}\\nmax : {}'\n",
    "    if x.dtype == ak.float64:\n",
    "        fmt = fmt.format(*['{:.2f}' for _ in range(4)])\n",
    "    print(fmt.format(x.mean(), x.std(), x.min(), x.max()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "describe(ak_green['fare_amount'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Histograms"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from matplotlib import pyplot as plt\n",
    "\n",
    "def hist(x, bins, log=True):\n",
    "    assert bins > 0\n",
    "    # Compute histogram counts in arkouda\n",
    "    h = ak.histogram(x, bins)\n",
    "    # Compute bins in numpy\n",
    "    if isinstance(x, ak.Datetime):\n",
    "        # Matplotlib has trouble plotting np.datetime64 and np.timedelta64\n",
    "        bins = ak.date_range(x.min(), x.max(), periods=bins).to_ndarray().astype('int')\n",
    "    elif isinstance(x, ak.Timedelta):\n",
    "        bins = ak.timedelta_range(x.min(), x.max(), periods=bins).to_ndarray().astype('int')\n",
    "    else:\n",
    "        bins = np.linspace(x.min(), x.max(), bins+1)[:-1]\n",
    "    # Bring h over to numpy for plotting\n",
    "    plt.bar(bins, h.to_ndarray(), width=bins[1]-bins[0])\n",
    "    if log:\n",
    "        plt.yscale('log')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hist(ak_green['fare_amount'], 100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Logical Indexing (Filters)\n",
    "Find non-negative fares"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "nonneg = ak_green['fare_amount'] >= 0\n",
    "print(f'{nonneg.sum() / nonneg.size :.1%} of fares are non-negative')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Select only non-negative fares for computation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "describe(ak_green['fare_amount'][nonneg])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Make new data dict with only non-negative fares"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_nonneg = {k:v[nonneg] for k, v in ak_green.items()}\n",
    "data_nonneg"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Time Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO - add new column once append is merged\n",
    "ride_duration = ak_green['lpep_dropoff_datetime'] - ak_green['lpep_pickup_datetime']\n",
    "ride_duration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ride_duration.min(), ride_duration.max()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hist(ride_duration, 100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Taxi Zone Lookup Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def cvt_to_string(v):\n",
    "    try:\n",
    "        if v == '':\n",
    "            return 'N/A'\n",
    "        else:\n",
    "            return str(v)\n",
    "    except:\n",
    "        return 'N/A'\n",
    "\n",
    "# read the taxi-zone-lookup-table\n",
    "cvt = {'Borough':cvt_to_string, 'Zone':cvt_to_string, 'service_zone':cvt_to_string}\n",
    "tzlut = pd.read_csv(\"/Users/ethandebandi/Documents/test_data/taxi+_zone_lookup.csv\",converters=cvt)\n",
    "\n",
    "# TODO - use ak.DataFrame once concat is merged\n",
    "\n",
    "# location id is 1-based, index is 0-based\n",
    "# fix it up to be aligned with index in data frame\n",
    "# which means add row zero\n",
    "top_row = pd.DataFrame({'LocationID': [0], 'Borough': ['N/A'], 'Zone': ['N/A'], 'service_zone': ['N/A']})\n",
    "tzlut = pd.concat([top_row, tzlut]).reset_index(drop = True)\n",
    "\n",
    "ak_tzlut = ak_from_pandas(tzlut)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ak_tzlut"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Apply Lookup Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "(ak_tzlut['LocationID'] == ak.arange(ak_tzlut['LocationID'].size)).all()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO - add column to ak_green once concat/append merged to codebase\n",
    "pu_borough = ak_tzlut['Borough'][ak_green['PULocationID']]\n",
    "do_borough = ak_tzlut['Borough'][ak_green['DOLocationID']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO - add column to ak_green once concat/append merged to codebase\n",
    "pu_zone = ak_tzlut['Zone'][ak_green['PULocationID']]\n",
    "do_zone = ak_tzlut['Zone'][ak_green['DOLocationID']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO - print ak_green with data appended\n",
    "pu_borough"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## GroupBy: Construct a Graph\n",
    "\n",
    "Directed graph from PULocationID --> DOLocationID"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "byloc = ak.GroupBy([ak_green['PULocationID'], ak_green['DOLocationID']])\n",
    "byloc.unique_keys"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Edge weight is number of rides\n",
    "\n",
    "Aggregation methods of `GroupBy` return tuple of (unique_keys, aggregate_values)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "(u, v), w = byloc.count()\n",
    "u, v, w"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Broadcast: Find Rides with Anomalous Fares\n",
    "\n",
    "Compute mean and std of fare by (pickup, dropoff)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "_, mf = byloc.mean(ak_green['fare_amount'])\n",
    "\n",
    "sf = (byloc.sum(ak_green['fare_amount']**2)[1] / w) - mf**2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Broadcast group values back to ride dataframe to compute z-scores of rides"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO - add column to ak_green once concat/append merged to codebase\n",
    "fare_mean = byloc.broadcast(mf, permute=True)\n",
    "fare_std = byloc.broadcast(sf, permute=True)\n",
    "\n",
    "fare_z = (ak_green['fare_amount'] - fare_mean) / (fare_std + 1)\n",
    "\n",
    "hist(fare_z, 100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Bring Small Result Set Back to Pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO - add once data is all merged in ak_green. See NYCTaxi_small.ipynb"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Disconnect from the server or shutdown the server"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#ak.disconnect()\n",
    "ak.shutdown()"
   ]
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "c9843ec7bf92a92e5648c61892a56b72546cb6f7f783cc4fa72865bc7befe91a"
  },
  "kernelspec": {
   "display_name": "Python 3.9.7 ('arkouda')",
   "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.9.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}