ELT_source_to_delta_table.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "d4c85537-af97-4dc5-9a44-2bd70acc9f01",
   "metadata": {},
   "outputs": [],
   "source": [
    "#!pip install deltalake\n",
    "#!pip install duckdb\n",
    "#!pip install pandas\n",
    "#!pip install numpy"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "cacb313b-bd8a-45e2-a25c-92f523a47aae",
   "metadata": {},
   "outputs": [],
   "source": [
    "import duckdb\n",
    "from deltalake.writer import write_deltalake\n",
    "from deltalake import DeltaTable\n",
    "from deltalake import DataCatalog"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "7ba145bc-e40f-45fc-aaa4-403b9078286c",
   "metadata": {},
   "outputs": [],
   "source": [
    "aws_access_key_id=\"\"\n",
    "aws_secret_access_key=\"\"\n",
    "aws_session_token=\"\"\n",
    "aws_region=\"\"\n",
    "src_s3_uri=\"s3://\"\n",
    "target_s3_uri=\"s3://\"\n",
    "table_name = \"tbl_nyc_taxi_trips\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "b887a05f-e7e8-446d-a342-919d57de736c",
   "metadata": {},
   "outputs": [],
   "source": [
    "storage_options = {\n",
    "    \"AWS_REGION\": aws_region,\n",
    "    'AWS_ACCESS_KEY_ID': aws_access_key_id,\n",
    "    'AWS_SECRET_ACCESS_KEY': aws_secret_access_key,\n",
    "    'AWS_SESSION_TOKEN': aws_session_token,\n",
    "    'AWS_S3_ALLOW_UNSAFE_RENAME': 'true'\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "aec160d4-c477-4693-99b3-dc12c1138629",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "┌─────────┐\n",
      "│ Success │\n",
      "│ boolean │\n",
      "├─────────┤\n",
      "│ true    │\n",
      "└─────────┘\n",
      "\n"
     ]
    }
   ],
   "source": [
    "create_secret_ddl = f\"\"\"\n",
    "    CREATE OR REPLACE SECRET my_ephem_sts_token \n",
    "        ( \n",
    "            TYPE S3,\n",
    "            KEY_ID '{aws_access_key_id}',\n",
    "            SECRET '{aws_secret_access_key}',\n",
    "            SESSION_TOKEN '{aws_session_token}',\n",
    "            REGION '{aws_region}'\n",
    "        );\"\"\"\n",
    "\n",
    "duckdb.sql(create_secret_ddl).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "b61d9a7f-017c-4b3f-b24d-a25996f0df7f",
   "metadata": {},
   "outputs": [],
   "source": [
    "src_green_sql = f\"\"\"\n",
    "    SELECT\n",
    "      gr.vendorid,\n",
    "      gr.lpep_pickup_datetime as pickup_datetime,\n",
    "      gr.lpep_dropoff_datetime as dropoff_datetime,\n",
    "      gr.ratecodeid,\n",
    "      gr.PULocationID as pickup_location_id,\n",
    "      gr.DOLocationID as drop_off_location_id,\n",
    "      gr.passenger_count,\n",
    "      gr.trip_distance,\n",
    "      gr.fare_amount,\n",
    "      gr.extra,\n",
    "      gr.mta_tax,\n",
    "      gr.tip_amount,\n",
    "      gr.tolls_amount,\n",
    "      gr.ehail_fee,\n",
    "      gr.improvement_surcharge,\n",
    "      gr.total_amount,\n",
    "      gr.payment_type,\n",
    "      gr.trip_type,\n",
    "      gr.congestion_surcharge,\n",
    "      0.0 as airport_fee,\n",
    "      'green' as cab_type,\n",
    "      '2024' as trip_year,\n",
    "      'January' as trip_month\n",
    "    FROM \n",
    "      '{src_s3_uri}/green_tripdata_2024-01.parquet' as gr;\n",
    "\"\"\"\n",
    "\n",
    "jan_green_cab_trips = duckdb.sql(src_green_sql)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "0d4ab9ed-624a-495a-9f2b-022552c79080",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 56551 entries, 0 to 56550\n",
      "Data columns (total 23 columns):\n",
      " #   Column                 Non-Null Count  Dtype         \n",
      "---  ------                 --------------  -----         \n",
      " 0   VendorID               56551 non-null  int32         \n",
      " 1   pickup_datetime        56551 non-null  datetime64[us]\n",
      " 2   dropoff_datetime       56551 non-null  datetime64[us]\n",
      " 3   RatecodeID             53136 non-null  float64       \n",
      " 4   pickup_location_id     56551 non-null  int32         \n",
      " 5   drop_off_location_id   56551 non-null  int32         \n",
      " 6   passenger_count        53136 non-null  float64       \n",
      " 7   trip_distance          56551 non-null  float64       \n",
      " 8   fare_amount            56551 non-null  float64       \n",
      " 9   extra                  56551 non-null  float64       \n",
      " 10  mta_tax                56551 non-null  float64       \n",
      " 11  tip_amount             56551 non-null  float64       \n",
      " 12  tolls_amount           56551 non-null  float64       \n",
      " 13  ehail_fee              0 non-null      float64       \n",
      " 14  improvement_surcharge  56551 non-null  float64       \n",
      " 15  total_amount           56551 non-null  float64       \n",
      " 16  payment_type           53136 non-null  float64       \n",
      " 17  trip_type              53133 non-null  float64       \n",
      " 18  congestion_surcharge   53136 non-null  float64       \n",
      " 19  airport_fee            56551 non-null  float64       \n",
      " 20  cab_type               56551 non-null  object        \n",
      " 21  trip_year              56551 non-null  object        \n",
      " 22  trip_month             56551 non-null  object        \n",
      "dtypes: datetime64[us](2), float64(15), int32(3), object(3)\n",
      "memory usage: 9.3+ MB\n"
     ]
    }
   ],
   "source": [
    "# test AWS connection\n",
    "jan_green_cab_trips.df().info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "1173cf23-3f5b-4eb5-bcf3-9f1b82669aaa",
   "metadata": {},
   "outputs": [],
   "source": [
    "write_deltalake(\n",
    "    table_or_uri=f\"{target_s3_uri}/{table_name}\",\n",
    "    storage_options=storage_options,\n",
    "    data=jan_green_cab_trips.df(),\n",
    "    mode='append',\n",
    "    partition_by=[\"trip_month\", \"trip_year\", \"cab_type\"], # only required for the first write, i.e. table creation with initial data\n",
    "    name=table_name\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba48bcf3-4a00-423d-873c-118cff17fc25",
   "metadata": {},
   "source": [
    "### Load 2nd file with Yellow Taxi Data for Jan, 2024"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "751f3da3-8be8-4732-b451-ad3acc8b13e7",
   "metadata": {},
   "outputs": [],
   "source": [
    "src_yellow_sql = f\"\"\"\n",
    "    SELECT\n",
    "      yl.vendorid,\n",
    "      yl.tpep_pickup_datetime as pickup_datetime,\n",
    "      yl.tpep_dropoff_datetime as dropoff_datetime,\n",
    "      yl.ratecodeid,\n",
    "      yl.PULocationID as pickup_location_id,\n",
    "      yl.DOLocationID as drop_off_location_id,\n",
    "      yl.passenger_count,\n",
    "      yl.trip_distance,\n",
    "      yl.fare_amount,\n",
    "      yl.extra,\n",
    "      yl.mta_tax,\n",
    "      yl.tip_amount,\n",
    "      yl.tolls_amount,\n",
    "      0.0 as ehail_fee,\n",
    "      yl.improvement_surcharge,\n",
    "      yl.total_amount,\n",
    "      yl.payment_type,\n",
    "      0 as trip_type,\n",
    "      yl.congestion_surcharge,\n",
    "      yl.Airport_fee as airport_fee,\n",
    "      'yellow' as cab_type,\n",
    "      '2024' as trip_year,\n",
    "      'January' as trip_month\n",
    "    FROM \n",
    "      '{src_s3_uri}/yellow_tripdata_2024-01.parquet' as yl;\n",
    "\"\"\"\n",
    "yellow_cab_trips = duckdb.sql(src_yellow_sql)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "f27f3616-8bbc-4e29-bbf5-9517055160cb",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 2964624 entries, 0 to 2964623\n",
      "Data columns (total 23 columns):\n",
      " #   Column                 Dtype         \n",
      "---  ------                 -----         \n",
      " 0   VendorID               int32         \n",
      " 1   pickup_datetime        datetime64[us]\n",
      " 2   dropoff_datetime       datetime64[us]\n",
      " 3   RatecodeID             float64       \n",
      " 4   pickup_location_id     int32         \n",
      " 5   drop_off_location_id   int32         \n",
      " 6   passenger_count        float64       \n",
      " 7   trip_distance          float64       \n",
      " 8   fare_amount            float64       \n",
      " 9   extra                  float64       \n",
      " 10  mta_tax                float64       \n",
      " 11  tip_amount             float64       \n",
      " 12  tolls_amount           float64       \n",
      " 13  ehail_fee              float64       \n",
      " 14  improvement_surcharge  float64       \n",
      " 15  total_amount           float64       \n",
      " 16  payment_type           int64         \n",
      " 17  trip_type              int32         \n",
      " 18  congestion_surcharge   float64       \n",
      " 19  airport_fee            float64       \n",
      " 20  cab_type               object        \n",
      " 21  trip_year              object        \n",
      " 22  trip_month             object        \n",
      "dtypes: datetime64[us](2), float64(13), int32(4), int64(1), object(3)\n",
      "memory usage: 475.0+ MB\n"
     ]
    }
   ],
   "source": [
    "# test AWS connection\n",
    "yellow_cab_trips.df().info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "478d0738-8361-4c7f-a35a-007d467a5b8f",
   "metadata": {},
   "outputs": [],
   "source": [
    "write_deltalake(\n",
    "    table_or_uri=f\"{target_s3_uri}/{table_name}\",\n",
    "    storage_options=storage_options,\n",
    "    data=yellow_cab_trips.df(),\n",
    "    mode='append',\n",
    "    name=table_name\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dd36daad-9ef1-461f-8ccd-a4fd52caf09f",
   "metadata": {},
   "source": [
    "## Load Green & Yellow for Feb 2024"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "dfcf6b33-2d6e-43fd-88d5-6b22892451cf",
   "metadata": {},
   "outputs": [],
   "source": [
    "src_yellow_feb_sql = f\"\"\"\n",
    "    SELECT\n",
    "      yl.vendorid,\n",
    "      yl.tpep_pickup_datetime as pickup_datetime,\n",
    "      yl.tpep_dropoff_datetime as dropoff_datetime,\n",
    "      yl.ratecodeid,\n",
    "      yl.PULocationID as pickup_location_id,\n",
    "      yl.DOLocationID as drop_off_location_id,\n",
    "      yl.passenger_count,\n",
    "      yl.trip_distance,\n",
    "      yl.fare_amount,\n",
    "      yl.extra,\n",
    "      yl.mta_tax,\n",
    "      yl.tip_amount,\n",
    "      yl.tolls_amount,\n",
    "      0.0 as ehail_fee,\n",
    "      yl.improvement_surcharge,\n",
    "      yl.total_amount,\n",
    "      yl.payment_type,\n",
    "      0 as trip_type,\n",
    "      yl.congestion_surcharge,\n",
    "      yl.Airport_fee as airport_fee,\n",
    "      'yellow' as cab_type,\n",
    "      '2024' as trip_year,\n",
    "      'February' as trip_month\n",
    "    FROM \n",
    "      '{src_s3_uri}/yellow_tripdata_2024-02.parquet' as yl;\n",
    "\"\"\"\n",
    "feb_yellow_cab_trips = duckdb.sql(src_yellow_feb_sql)\n",
    "\n",
    "src_green_feb_sql = f\"\"\"\n",
    "    SELECT\n",
    "      gr.vendorid,\n",
    "      gr.lpep_pickup_datetime as pickup_datetime,\n",
    "      gr.lpep_dropoff_datetime as dropoff_datetime,\n",
    "      gr.ratecodeid,\n",
    "      gr.PULocationID as pickup_location_id,\n",
    "      gr.DOLocationID as drop_off_location_id,\n",
    "      gr.passenger_count,\n",
    "      gr.trip_distance,\n",
    "      gr.fare_amount,\n",
    "      gr.extra,\n",
    "      gr.mta_tax,\n",
    "      gr.tip_amount,\n",
    "      gr.tolls_amount,\n",
    "      gr.ehail_fee,\n",
    "      gr.improvement_surcharge,\n",
    "      gr.total_amount,\n",
    "      gr.payment_type,\n",
    "      gr.trip_type,\n",
    "      gr.congestion_surcharge,\n",
    "      0.0 as airport_fee,\n",
    "      'green' as cab_type,\n",
    "      '2024' as trip_year,\n",
    "      'February' as trip_month\n",
    "    FROM \n",
    "      '{src_s3_uri}/green_tripdata_2024-02.parquet' as gr;\n",
    "\"\"\"\n",
    "\n",
    "feb_green_cab_trips = duckdb.sql(src_green_feb_sql)\n",
    "\n",
    "gr_and_yl_feb_sql = \"\"\"\n",
    "    SELECT * from feb_green_cab_trips\n",
    "    UNION\n",
    "    SELECT * from feb_yellow_cab_trips\n",
    "\"\"\"\n",
    "\n",
    "feb_cab_trips = duckdb.sql(gr_and_yl_feb_sql)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "6ca4116d-0505-4269-86c8-2fd4bdb5e8b5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pandas.core.frame.DataFrame'>\n",
      "RangeIndex: 3061102 entries, 0 to 3061101\n",
      "Data columns (total 23 columns):\n",
      " #   Column                 Dtype         \n",
      "---  ------                 -----         \n",
      " 0   VendorID               int32         \n",
      " 1   pickup_datetime        datetime64[us]\n",
      " 2   dropoff_datetime       datetime64[us]\n",
      " 3   RatecodeID             float64       \n",
      " 4   pickup_location_id     int32         \n",
      " 5   drop_off_location_id   int32         \n",
      " 6   passenger_count        float64       \n",
      " 7   trip_distance          float64       \n",
      " 8   fare_amount            float64       \n",
      " 9   extra                  float64       \n",
      " 10  mta_tax                float64       \n",
      " 11  tip_amount             float64       \n",
      " 12  tolls_amount           float64       \n",
      " 13  ehail_fee              float64       \n",
      " 14  improvement_surcharge  float64       \n",
      " 15  total_amount           float64       \n",
      " 16  payment_type           float64       \n",
      " 17  trip_type              float64       \n",
      " 18  congestion_surcharge   float64       \n",
      " 19  airport_fee            float64       \n",
      " 20  cab_type               object        \n",
      " 21  trip_year              object        \n",
      " 22  trip_month             object        \n",
      "dtypes: datetime64[us](2), float64(15), int32(3), object(3)\n",
      "memory usage: 502.1+ MB\n"
     ]
    }
   ],
   "source": [
    "feb_cab_trips.df().info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "66729aa2-0cde-46b7-a43b-3567526fceb5",
   "metadata": {},
   "outputs": [],
   "source": [
    "write_deltalake(\n",
    "    table_or_uri=f\"{target_s3_uri}/{table_name}\",\n",
    "    storage_options=storage_options,\n",
    "    data=feb_cab_trips.df(),\n",
    "    mode='append',\n",
    "    name=table_name\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3c326d25-76b0-4051-8c6e-da78433171eb",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}