preprocess_db_setup_load_script.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "204373cc-5a15-4231-ae8f-8f95401164aa",
   "metadata": {},
   "source": [
    "# TPCDS: Preprocessing, DB Setup and Data Load Script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "06b7870c-b590-469b-998c-15e65b729bc8",
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys, os, re\n",
    "import psycopg2\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "from psycopg2 import Error\n",
    "from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "78ab3e3c-62bf-4d52-bc95-71c5b8f3efe0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# set up connection variables\n",
    "db_host = \"localhost\"\n",
    "db_port = \"5432\"\n",
    "db_user = \"postgres\"\n",
    "db_pass = \"password\"\n",
    "db_name = \"postgres\"\n",
    "\n",
    "# function to connect with postgres\n",
    "def connect_postgres(db_host, db_port, db_user, db_pass, db_name):\n",
    "    try:\n",
    "        # Connect to an existing database\n",
    "        connection = psycopg2.connect(host = db_host,\n",
    "                                      port = db_port,\n",
    "                                      user = db_user,\n",
    "                                      password = db_pass,\n",
    "                                      database = db_name)\n",
    "        # Set auto-commit\n",
    "        connection.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT);\n",
    "        # Create a cursor to perform database operations\n",
    "        cur = connection.cursor()\n",
    "        # Print PostgreSQL details\n",
    "        print(\"PostgreSQL server information\")\n",
    "        print(connection.get_dsn_parameters(), \"\\n\")\n",
    "        # Executing a SQL query\n",
    "        cur.execute(\"SELECT version();\")\n",
    "        # Fetch result\n",
    "        record = cur.fetchone()\n",
    "        print(\"You are connected to - \", record, \"\\n\")\n",
    "\n",
    "    except (Exception, Error) as error:\n",
    "        print(\"Error while connecting to PostgreSQL\", error)\n",
    "    else:\n",
    "        return cur"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "e924ce7a-d5ed-47bf-8251-aee4008ebed2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PostgreSQL server information\n",
      "{'user': 'postgres', 'dbname': 'postgres', 'host': 'localhost', 'port': '5432', 'tty': '', 'options': '', 'sslmode': 'prefer', 'sslcompression': '0', 'gssencmode': 'prefer', 'krbsrvname': 'postgres', 'target_session_attrs': 'any'} \n",
      "\n",
      "You are connected to -  ('PostgreSQL 14.5, compiled by Visual C++ build 1914, 64-bit',) \n",
      "\n"
     ]
    }
   ],
   "source": [
    "# connect to postgres\n",
    "\n",
    "cur = connect_postgres(db_host, db_port, db_user, db_pass, db_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "e82adf59-d8df-4f70-bc44-3d0d1398daca",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SQL Status Output:\n",
      " DROP DATABASE\n"
     ]
    }
   ],
   "source": [
    "# drop tpcds db\n",
    "\n",
    "db_name = \"tpcds\"\n",
    "\n",
    "cur.execute(\n",
    "    f\"DROP DATABASE IF EXISTS {db_name} WITH (FORCE);\"\n",
    ")\n",
    "print(\"SQL Status Output:\\n\", cur.statusmessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "025c2e12-59cd-4380-b22b-285d1edd7bc5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SQL Status Output:\n",
      " ALTER DATABASE\n"
     ]
    }
   ],
   "source": [
    "# change win1252 encoding temp db to normal before drop\n",
    "try:\n",
    "    cur.execute(\n",
    "        \"ALTER DATABASE win1252_temp is_template false;\"\n",
    "    )\n",
    "except Exception as e:\n",
    "    print(e)\n",
    "else:\n",
    "    print(\"SQL Status Output:\\n\", cur.statusmessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "9bce6df0-62d3-4257-a534-2cdb8dc2a75e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SQL Status Output:\n",
      " DROP DATABASE\n"
     ]
    }
   ],
   "source": [
    "# drop win1252 encoding temp db (after set to normal db)\n",
    "cur.execute(\n",
    "    \"DROP DATABASE IF EXISTS win1252_temp WITH (FORCE);\"\n",
    ")\n",
    "print(\"SQL Status Output:\\n\", cur.statusmessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "51426edd-2e89-432e-8adc-2c96a958b667",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SQL Status Output:\n",
      " CREATE DATABASE\n"
     ]
    }
   ],
   "source": [
    "# create win1252 encoding temp db\n",
    "cur.execute(\n",
    "    \"\"\"\n",
    "    \n",
    "    CREATE DATABASE win1252_temp\n",
    "        WITH\n",
    "        OWNER = postgres\n",
    "        TEMPLATE = template0\n",
    "        ENCODING = 'WIN1252'\n",
    "        CONNECTION LIMIT = -1\n",
    "        IS_TEMPLATE = True;\n",
    "\n",
    "    \"\"\"\n",
    ")\n",
    "print(\"SQL Status Output:\\n\", cur.statusmessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "87d07b64-0624-4b39-b361-78ab1e318be0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SQL Status Output:\n",
      " CREATE DATABASE\n"
     ]
    }
   ],
   "source": [
    "# create tpcds db\n",
    "cur.execute(\n",
    "    f\"\"\"\n",
    "\n",
    "    CREATE DATABASE {db_name}\n",
    "        WITH\n",
    "        OWNER = postgres\n",
    "        TEMPLATE = win1252_temp\n",
    "        ENCODING = 'WIN1252'\n",
    "        CONNECTION LIMIT = -1\n",
    "        IS_TEMPLATE = False;\n",
    "        \n",
    "    \"\"\"\n",
    ")\n",
    "print(\"SQL Status Output:\\n\", cur.statusmessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "9758b305-b016-4cf7-80a4-ede8cf6b3db0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PostgreSQL server information\n",
      "{'user': 'postgres', 'dbname': 'tpcds', 'host': 'localhost', 'port': '5432', 'tty': '', 'options': '', 'sslmode': 'prefer', 'sslcompression': '0', 'gssencmode': 'prefer', 'krbsrvname': 'postgres', 'target_session_attrs': 'any'} \n",
      "\n",
      "You are connected to -  ('PostgreSQL 14.5, compiled by Visual C++ build 1914, 64-bit',) \n",
      "\n"
     ]
    }
   ],
   "source": [
    "# connect to tpcds db\n",
    "\n",
    "cur = connect_postgres(db_host, db_port, db_user, db_pass, db_name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "16bf6841-d1a7-41f0-82b5-b58a8ab793f5",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SQL Status Output:\n",
      " CREATE TABLE\n",
      "SQL Status Output:\n",
      " CREATE TABLE\n"
     ]
    }
   ],
   "source": [
    "# create tables for db\n",
    "\n",
    "cur.execute(open(\"tools/tpcds.sql\", \"r\").read())\n",
    "print(\"SQL Status Output:\\n\", cur.statusmessage)\n",
    "cur.execute(open(\"tools/tpcds_source.sql\", \"r\").read())\n",
    "print(\"SQL Status Output:\\n\", cur.statusmessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "77146980-b68a-4d14-9e03-2c9b2029d767",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "C:\\Users\\ahmad\\Desktop\\tpc_ds\\tools\\tmp\\\n"
     ]
    }
   ],
   "source": [
    "# get dir path\n",
    "\n",
    "path = os.getcwd() + '\\\\tools\\\\tmp\\\\'\n",
    "files = os.listdir(path)\n",
    "print(path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "273d8a93-d9d5-4820-8734-5cabf3d9e1c6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# function to get full abosolute path of csv files containing data\n",
    "\n",
    "def get_absolute_path(d):\n",
    "    return [os.path.join(d, f) for f in os.listdir(d)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "c101c6c8-ec5e-40d3-a94b-678f86ef9076",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total files: 33\n",
      "First few files...\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "['C:/Users/ahmad/Desktop/tpc_ds/tools/tmp/call_center_1_5.csv',\n",
       " 'C:/Users/ahmad/Desktop/tpc_ds/tools/tmp/catalog_page_1_5.csv',\n",
       " 'C:/Users/ahmad/Desktop/tpc_ds/tools/tmp/catalog_returns_1_5.csv',\n",
       " 'C:/Users/ahmad/Desktop/tpc_ds/tools/tmp/catalog_sales_1_5.csv',\n",
       " 'C:/Users/ahmad/Desktop/tpc_ds/tools/tmp/customer_1_5.csv']"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# get full abosolute path of csv files containing data\n",
    "\n",
    "files_abs_path = [p.replace('\\\\', '/') for p in get_absolute_path(path)]\n",
    "print(\"Total files:\", len(files_abs_path))\n",
    "print(\"First few files...\")\n",
    "files_abs_path[:5]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "c50b4819-73ce-48cb-8e60-2647adb46727",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Iteration 1 done!\n",
      "1 file(s) updated for extra column exclusion.\n"
     ]
    }
   ],
   "source": [
    "# exclude extra delimiter for dbgen_version file\n",
    "\n",
    "file_count = 0\n",
    "\n",
    "for iteration in range(0, 1):\n",
    "    for file in files_abs_path:\n",
    "        file_open = open(file,'r')  \n",
    "        all_text = file_open.read().replace(\" \", \"\")\n",
    "        file_open.close()\n",
    "\n",
    "        if (all_text[-13] == '^' and 'dbgen_version' in file):\n",
    "            file_open_read = open(file, 'r', encoding = 'latin-1')\n",
    "            string_list = file_open_read.readlines()\n",
    "            file_open_read.close()\n",
    "\n",
    "            for i in range(len(string_list)):\n",
    "                last_delimeter_index = string_list[i].rfind(\"^\")\n",
    "                string_list[i] = string_list[i][:last_delimeter_index] + \"\" + string_list[i][last_delimeter_index + 1:]\n",
    "\n",
    "            file_open_write = open(file, 'w', encoding = 'latin-1')\n",
    "            new_file_contents = ''.join(string_list)\n",
    "            file_open_write.write(new_file_contents)\n",
    "            file_open_write.close()\n",
    "            \n",
    "            file_count += 1\n",
    "        else:\n",
    "            pass\n",
    "    print(f'\\nIteration {iteration + 1} done!')\n",
    "    print(f'{file_count} file(s) updated for extra column exclusion.')\n",
    "    file_count = 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "9fadd633-a4b6-4132-ba36-edc99affaa64",
   "metadata": {},
   "outputs": [],
   "source": [
    "# generate sql commands for loading data from csv to postgres db\n",
    "# considers that csv files were generated in parallel stream\n",
    "\n",
    "sql_commands_file = open('data_load_script.sql','w')\n",
    "\n",
    "for file in files:\n",
    "    underscore_index = [underscore_ind.start() for underscore_ind in re.finditer('_', file)]\n",
    "    file_name = file[:underscore_index[-2]]\n",
    "    file_path = path+file\n",
    "    sql_command = \"COPY public.\"+file_name+\" FROM '\"+file_path+\"' delimiter '^' CSV;\\n\"\n",
    "    sql_commands_file.write(sql_command)\n",
    "\n",
    "sql_commands_file.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "8d91df03-0169-4cc2-a179-5796470ff765",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SQL Status Output:\n",
      " COPY 30\n"
     ]
    }
   ],
   "source": [
    "# load csv data into db\n",
    "\n",
    "cur.execute(open(\"data_load_script.sql\", \"r\").read())\n",
    "print(\"SQL Status Output:\\n\", cur.statusmessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "f47515db-2649-48fa-8109-f31c246aca64",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "SQL Status Output:\n",
      " ALTER TABLE\n"
     ]
    }
   ],
   "source": [
    "# add constraints to db\n",
    "\n",
    "cur.execute(open(\"tools/tpcds_ri.sql\", \"r\").read())\n",
    "print(\"SQL Status Output:\\n\", cur.statusmessage)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "f0d52bc3-9907-4fed-81b0-fb0cc196cafe",
   "metadata": {},
   "outputs": [],
   "source": [
    "# close connection to db\n",
    "\n",
    "cur.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e1e9cc6c-0819-4a8d-bf9a-5b26b7611b0c",
   "metadata": {},
   "source": [
    "#### End of script."
   ]
  }
 ],
 "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.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}