notebooks/12_simulated_feature_pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 📕 12 - Data Pipeline for Fetching New Data\n",
    "\n",
    "## Introduction\n",
    "\n",
    "Given that the New York City Taxi Association updates their data only on a monthly basis, we face a barrier in obtaining real-time data. To ensure that our predictive models remain current and our analytics dashboard reflects 'real-time' trends, I've devised a strategy. This involves simulating real-time data fetching by repurposing older datasets, making them appear as recent entries.\n",
    "\n",
    "Initially, I set a timeframe spanning the last 28 days from the current UTC time. This not only delineates a clear period of interest but also provides redundancy. Subsequently, I leverage the already existing `fetch_batch_raw_data` function. What it essentially does is retrieve data from a year (52 weeks) back, within our defined window, and then adjusts the timestamps by a year forward. This clever trick makes past data seem current. Once transformed into a time-series format, the data is seamlessly inserted into our Hopsworks feature group.\n",
    "\n",
    "It's crucial to emphasize that this approach simulates what would happen if we had direct access to a data warehouse. As part of the project workflow, we intend to utilize GitHub Actions to automate the data fetch process every hour.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import libraries\n",
    "from datetime import datetime, timedelta\n",
    "import pandas as pd\n",
    "import hopsworks\n",
    "import src.config as config\n",
    "from src.data import load_raw_data\n",
    "from src.data import transform_raw_data_into_ts_data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the current UTC time, rounded down to the nearest hour\n",
    "current_time = pd.to_datetime(datetime.utcnow()).floor(\"H\")\n",
    "    \n",
    "# Calculate the start of the timeframe\n",
    "from_date = current_time - timedelta(days=28)\n",
    "    \n",
    "# Set the end of the timeframe as the current time\n",
    "to_date = current_time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def fetch_batch_raw_data(from_date: datetime, to_date: datetime) -> pd.DataFrame:\n",
    "    \"\"\"\n",
    "    Simulate fetching recent batch data by repurposing older historical data.\n",
    "\n",
    "    This function fetches raw data from exactly 52 weeks (1 year) ago from the given date range, \n",
    "    then shifts the data timestamps by 52 weeks to simulate as if the data is recent.\n",
    "\n",
    "    Parameters:\n",
    "    - from_date (datetime): The start date of the required data range.\n",
    "    - to_date (datetime): The end date of the required data range.\n",
    "\n",
    "    Returns:\n",
    "    - pd.DataFrame: A DataFrame containing the simulated recent data.\n",
    "    \"\"\"\n",
    "    \n",
    "    # Calculate equivalent date range from a year ago\n",
    "    from_date_ = from_date - timedelta(days=7*52)\n",
    "    to_date_ = to_date - timedelta(days=7*52)\n",
    "\n",
    "    print(f\"Fetching raw data from {from_date_} to {to_date_}\")\n",
    "\n",
    "    # Load raw data for the old date range\n",
    "    rides = load_raw_data(year=from_date_.year, months=from_date_.month)\n",
    "    rides = rides[rides[\"pickup_datetime\"] >= from_date_]\n",
    "    \n",
    "    rides_2 = load_raw_data(year=to_date_.year, months=to_date_.month)\n",
    "    rides_2 = rides_2[rides_2[\"pickup_datetime\"] < to_date_]\n",
    "\n",
    "    # Combine both dataframes\n",
    "    rides = pd.concat([rides, rides_2])\n",
    "\n",
    "    # Shift the data by 52 weeks to make it look recent\n",
    "    rides[\"pickup_datetime\"] += timedelta(days=7*52)\n",
    "    \n",
    "    # Sort the dataframe by location and datetime\n",
    "    rides.sort_values(by=[\"pickup_location_id\", \"pickup_datetime\"], inplace=True)\n",
    "\n",
    "    return rides\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Fetching raw data from 2022-09-07 12:00:00 to 2022-10-05 12:00:00\n",
      "File 2022-09 was already in local storage\n",
      "File 2022-10 was already in local storage\n"
     ]
    }
   ],
   "source": [
    "# fetch batch raw data within the given 28 days timeframe\n",
    "rides = fetch_batch_raw_data(from_date, to_date)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "DeprecationWarning: np.find_common_type is deprecated.  Please use `np.result_type` or `np.promote_types`.\n",
      "See https://numpy.org/devdocs/release/1.25.0-notes.html and the docs for more information.  (Deprecated NumPy 1.25)\n",
      "DeprecationWarning: np.find_common_type is deprecated.  Please use `np.result_type` or `np.promote_types`.\n",
      "See https://numpy.org/devdocs/release/1.25.0-notes.html and the docs for more information.  (Deprecated NumPy 1.25)\n",
      "100%|██████████| 265/265 [00:00<00:00, 406.31it/s]\n"
     ]
    },
    {
     "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>pickup_hour</th>\n",
       "      <th>rides</th>\n",
       "      <th>pickup_location_id</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2023-09-06 12:00:00</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2023-09-06 13:00:00</td>\n",
       "      <td>2</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2023-09-06 14:00:00</td>\n",
       "      <td>4</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2023-09-06 15:00:00</td>\n",
       "      <td>1</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2023-09-06 16:00:00</td>\n",
       "      <td>0</td>\n",
       "      <td>1</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "          pickup_hour  rides  pickup_location_id\n",
       "0 2023-09-06 12:00:00      1                   1\n",
       "1 2023-09-06 13:00:00      2                   1\n",
       "2 2023-09-06 14:00:00      4                   1\n",
       "3 2023-09-06 15:00:00      1                   1\n",
       "4 2023-09-06 16:00:00      0                   1"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# convert raw data into time series data\n",
    "ts_data = transform_raw_data_into_ts_data(rides)\n",
    "ts_data.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Connection closed.\n",
      "Connected. Call `.close()` to terminate connection gracefully.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Logged in to project, explore it here https://c.app.hopsworks.ai:443/p/111769\n",
      "Connected. Call `.close()` to terminate connection gracefully.\n"
     ]
    }
   ],
   "source": [
    "# connect to the project\n",
    "project = hopsworks.login(\n",
    "    project=config.HOPSWORKS_PROJECT_NAME,\n",
    "    api_key_value=config.HOPSWORKS_API_KEY\n",
    ")\n",
    "\n",
    "# connect to the feature store\n",
    "feature_store = project.get_feature_store()\n",
    "\n",
    "# connect to the feature group\n",
    "feature_group = feature_store.get_or_create_feature_group(\n",
    "    name=config.FEATURE_GROUP_NAME,\n",
    "    version=config.FEATURE_GROUP_VERSION,\n",
    "    description=\"Time series data aggregated by hour\",\n",
    "    primary_key=[\"pickup_location_id\", \"pickup_hour\"],\n",
    "    event_time=\"pickup_hour\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "2ea32157156b425dbaadc0c575c5cdf4",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "Uploading Dataframe: 0.00% |          | Rows 0/178080 | Elapsed Time: 00:00 | Remaining Time: ?"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Launching job: time_series_hourly_feature_group_1_offline_fg_backfill\n",
      "Job started successfully, you can follow the progress at \n",
      "https://c.app.hopsworks.ai/p/111769/jobs/named/time_series_hourly_feature_group_1_offline_fg_backfill/executions\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "(<hsfs.core.job.Job at 0x1439021d0>, None)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# insert the simulated time series data into the feature group\n",
    "feature_group.insert(ts_data, write_options={\"wait_for_job\": False})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "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.10.0"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}