NYCTaxi_small.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",
    "2 methods:\n",
    "* Load in python (e.g. with pandas) and transfer to arkouda\n",
    "  * Good for prototyping with small data\n",
    "* Convert to HDF5 (in separate process) and read directly with arkouda\n",
    "  * Best for large datasets\n",
    "  \n",
    "#### Dtypes Supported in HDF5\n",
    "Arkouda can only read these HDF5 dtypes\n",
    "* int (any width) -> int64\n",
    "* float (any width) -> float64\n",
    "* custom string format -> ak.Strings\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 /home/reusters/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": [
    "### Method 1: CSV --> Pandas --> Arkouda"
   ]
  },
  {
   "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('/home/reusters/data/green_tripdata_2020-01.csv', **taxi.OPTIONS['green'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# transfer columns of DataFrame to arkouda\n",
    "def ak_create_akdict_from_df(df):\n",
    "    akdict = {}\n",
    "    for cname in df.keys():\n",
    "        if df[cname].dtype.name == 'object':\n",
    "            akdict[cname] = ak.from_series(df[cname],dtype=np.str)\n",
    "        else:\n",
    "            akdict[cname] = ak.from_series(df[cname])\n",
    "\n",
    "    return akdict"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "green_from_pandas = ak_create_akdict_from_df(pdgreen)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Method 2: CSV --> HDF5 --> Arkouda\n",
    "\n",
    "Arkouda comes with a CSV to HDF5 converter in the repo. It uses the NYCTaxi_format.py file we defined above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!python3 /home/reusters/arkouda/converter/csv2hdf.py \\\n",
    "--formats-file=/home/reusters/ArkoudaNotebooks/NYCTaxi_format.py \\\n",
    "--format=green \\\n",
    "--outdir=/home/reusters/data/ \\\n",
    "/home/reusters/data/green_tripdata_2020-01.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Can skip this if already connected above\n",
    "import arkouda as ak\n",
    "ak.connect(connect_url=\"tcp://localhost:5555\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "green_from_HDF5 = ak.read_all('/home/reusters/data/green_tripdata_2020-01.hdf')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Same Result from Both Methods"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def frames_are_equal(a, b):\n",
    "    # Ensure same columns\n",
    "    if a.keys() != b.keys():\n",
    "        return False\n",
    "    # Ensure same column dtypes\n",
    "    if not all(a[k].dtype == b[k].dtype for k in a):\n",
    "        return False\n",
    "    # Compare column values\n",
    "    for k in a:\n",
    "        # Workaround until ak.isna() is implemented\n",
    "        # Because nan != nan\n",
    "        if a[k].dtype == ak.float64:\n",
    "            cmp = ak.cast(a[k], 'int64') == ak.cast(b[k], 'int64')\n",
    "        else:\n",
    "            cmp = a[k] == b[k]\n",
    "        if not cmp.all():\n",
    "            return False\n",
    "    return True"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "frames_are_equal(green_from_HDF5, green_from_pandas)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Proceed with just one, since they are equivalent\n",
    "data = green_from_HDF5"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Convert Columns to Specialized Dtypes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data['lpep_pickup_datetime'] = ak.Datetime(data['lpep_pickup_datetime'])\n",
    "data['lpep_dropoff_datetime'] = ak.Datetime(data['lpep_dropoff_datetime'])\n",
    "data['store_and_fwd_flag'] = (data['store_and_fwd_flag'] == 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "rows = data['VendorID'].size\n",
    "numbytes = sum(v.size*v.itemsize for v in data.values())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f'{rows:,} rows\\n{numbytes:,} bytes')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exploration"
   ]
  },
  {
   "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(data['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(data['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 = data['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(data['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 data.items()}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_nonneg"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Time Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data['ride_duration'] = data['lpep_dropoff_datetime'] - data['lpep_pickup_datetime']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data['ride_duration']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data['ride_duration'].min(), data['ride_duration'].max()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hist(data['ride_duration'], 100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Taxi Zone Lookup Table"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Use Method 1: CSV --> Pandas --> Arkouda"
   ]
  },
  {
   "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(\"/home/reusters/data/taxi+_zone_lookup.csv\",converters=cvt)\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)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tzlut"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Convert dataframe to dictionary of Arkouda arrays"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# convert data frame with strings and int64 data\n",
    "aktzlut = ak_create_akdict_from_df(tzlut)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "aktzlut"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Apply Lookup Table"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "(aktzlut['LocationID'] == ak.arange(aktzlut['LocationID'].size)).all()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data['PUBorough'] = aktzlut['Borough'][data['PULocationID']]\n",
    "data['DOBorough'] = aktzlut['Borough'][data['DOLocationID']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data['PUZone'] = aktzlut['Zone'][data['PULocationID']]\n",
    "data['DOZone'] = aktzlut['Zone'][data['DOLocationID']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data"
   ]
  },
  {
   "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([data['PULocationID'], data['DOLocationID']])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "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()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "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(data['fare_amount'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sf = (byloc.sum(data['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": [
    "data['fare_mean'] = byloc.broadcast(mf, permute=True)\n",
    "data['fare_std'] = byloc.broadcast(sf, permute=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data['fare_z'] = (data['fare_amount'] - data['fare_mean']) / (data['fare_std'] + 1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hist(data['fare_z'], 100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Bring Small Result Set Back to Pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "exorbitant = (data['fare_z'] > 5)\n",
    "exdf = pd.DataFrame({k: v[exorbitant].to_ndarray() for k, v in data.items()})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "exdf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "worst = data['fare_z'].argmax()\n",
    "{k:v[worst] for k, v in data.items()}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Disconnect from the server or shutdown the server"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# disconnect or shutdown the server\n",
    "#ak.disconnect()\n",
    "#ak.shutdown()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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.7.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}