ETL_create_database.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "491818de",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "491818de",
    "outputId": "afb2d5ed-737d-407a-bf2f-89eb4245f3bf"
   },
   "outputs": [],
   "source": [
    "import json\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "\n",
    "import re\n",
    "\n",
    "from sqlalchemy import create_engine\n",
    "import psycopg2\n",
    "from config import db_password\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "709bd0b6",
   "metadata": {
    "id": "709bd0b6"
   },
   "outputs": [],
   "source": [
    "#  Add the clean movie function that takes in the argument, \"movie\".\n",
    "def clean_movie(movie):\n",
    "    alt_titles = {}\n",
    "\n",
    "    # combine alternate titles into one list\n",
    "    for key in ['Also known as','Arabic','Cantonese','Chinese','French',\n",
    "                'Hangul','Hebrew','Hepburn','Japanese','Literally',\n",
    "                'Mandarin','McCune–Reischauer','Original title','Polish',\n",
    "                'Revised Romanization','Romanized','Russian',\n",
    "                'Simplified','Traditional','Yiddish']:\n",
    "        if key in movie:\n",
    "            alt_titles[key] = movie[key]\n",
    "            movie.pop(key)\n",
    "    if len(alt_titles) > 0:\n",
    "        movie['alt_titles'] = alt_titles\n",
    "\n",
    "    # merge column names\n",
    "    def change_column_name(old_name, new_name):\n",
    "        if old_name in movie:\n",
    "            movie[new_name] = movie.pop(old_name)\n",
    "\n",
    "    change_column_name('Adaptation by', 'Writer(s)')\n",
    "    change_column_name('Country of origin', 'Country')\n",
    "    change_column_name('Directed by', 'Director')\n",
    "    change_column_name('Distributed by', 'Distributor')\n",
    "    change_column_name('Edited by', 'Editor(s)')\n",
    "    change_column_name('Length', 'Running time')\n",
    "    change_column_name('Original release', 'Release date')\n",
    "    change_column_name('Music by', 'Composer(s)')\n",
    "    change_column_name('Produced by', 'Producer(s)')\n",
    "    change_column_name('Producer', 'Producer(s)')\n",
    "    change_column_name('Productioncompanies ', 'Production company(s)')\n",
    "    change_column_name('Productioncompany ', 'Production company(s)')\n",
    "    change_column_name('Released', 'Release Date')\n",
    "    change_column_name('Release Date', 'Release date')\n",
    "    change_column_name('Screen story by', 'Writer(s)')\n",
    "    change_column_name('Screenplay by', 'Writer(s)')\n",
    "    change_column_name('Story by', 'Writer(s)')\n",
    "    change_column_name('Theme music composer', 'Composer(s)')\n",
    "    change_column_name('Written by', 'Writer(s)')\n",
    "\n",
    "    return movie"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "ca1292ed",
   "metadata": {
    "colab": {
     "base_uri": "https://localhost:8080/"
    },
    "id": "ca1292ed",
    "outputId": "913fe8b3-96b4-4063-84d3-f476cf38a050"
   },
   "outputs": [],
   "source": [
    "# 1 Add the function that takes in three arguments;\n",
    "# Wikipedia data, Kaggle metadata, and MovieLens rating data (from Kaggle)\n",
    "file_dir = \"./\"\n",
    "\n",
    "def three_arguments_func():\n",
    "    # Read in the kaggle metadata and MovieLens ratings CSV files as Pandas DataFrames.\n",
    "    kaggle_metadata = pd.read_csv(f'{file_dir}movies_metadata.csv', low_memory=False)\n",
    "    ratings = pd.read_csv(f'{file_dir}ratings.csv')\n",
    "    \n",
    "\n",
    "    # Open and read the Wikipedia data JSON file.\n",
    "    with open(f'{file_dir}wikipedia-movies.json', mode='r') as file:\n",
    "        wiki_movies_raw = json.load(file)\n",
    "    \n",
    "    # 5. Return the three DataFrames\n",
    "    return wiki_movies_raw, kaggle_metadata, ratings\n",
    "\n",
    "\n",
    "# 7. Set the three variables in Step 6 equal to the function created in Step 1.\n",
    "def extract_transform_load():\n",
    "    #  2. Read in the kaggle metadata and MovieLens ratings CSV files as Pandas DataFrames.\n",
    "    wiki_movies_raw, kaggle_metadata, ratings = three_arguments_func()\n",
    "\n",
    "    # Write a list comprehension to filter out TV shows.\n",
    "    wiki_movies = [movie for movie in wiki_movies_raw if ('Director' in movie or 'Directed by' in movie) and 'imdb_link' in movie and 'No. of episodes' not in movie]\n",
    " \n",
    "    # Write a list comprehension to iterate through the cleaned wiki movies list\n",
    "    # and call the clean_movie function on each movie.\n",
    "    clean_movies = [clean_movie(movie) for movie in wiki_movies]     \n",
    "\n",
    "    # Read in the cleaned movies list from Step 4 as a DataFrame.\n",
    "    wiki_movies_df = pd.DataFrame(clean_movies)\n",
    "\n",
    "    # Write a try-except block to catch errors while extracting the IMDb ID using a regular expression string and\n",
    "    #  dropping any imdb_id duplicates. If there is an error, capture and print the exception.\n",
    "    try:\n",
    "        wiki_movies_df['imdb_id'] = wiki_movies_df['imdb_link'].str.extract(r'(tt\\d{7})')\n",
    "        wiki_movies_df.drop_duplicates(subset='imdb_id', inplace=True)\n",
    "    \n",
    "    except Exception as e: \n",
    "        print(e) \n",
    "\n",
    "    #  Write a list comprehension to keep the columns that don't have null values from the wiki_movies_df DataFrame.\n",
    "    wiki_columns_to_keep = [column for column in wiki_movies_df.columns if wiki_movies_df[column].isnull().sum() < len(wiki_movies_df) * 0.9]\n",
    "    wiki_movies_df = wiki_movies_df[wiki_columns_to_keep]    \n",
    "\n",
    "    # Create a variable that will hold the non-null values from the “Box office” column.\n",
    "    box_office = wiki_movies_df['Box office'].dropna()\n",
    "    \n",
    "    # Convert the box office data created in Step 8 to string values using the lambda and join functions.\n",
    "    box_office = box_office.apply(lambda x: ' '.join(x) if type(x) == list else x)   \n",
    "\n",
    "    # Write a regular expression to match the six elements of \"form_one\" of the box office data.\n",
    "    form_one = r'\\$\\s*\\d+\\.?\\d*\\s*[mb]illi?on'\n",
    "    \n",
    "    # Write a regular expression to match the three elements of \"form_two\" of the box office data.\n",
    "    form_two = r'\\$\\s*\\d{1,3}(?:[,\\.]\\d{3})+(?!\\s[mb]illion)'    \n",
    "\n",
    "    # Add the parse_dollars function.\n",
    "    def parse_dollars(s):\n",
    "        if type(s) != str:\n",
    "            return np.nan\n",
    "        if re.match(r'\\$\\s*\\d+\\.?\\d*\\s*milli?on', s, flags=re.IGNORECASE):\n",
    "            s = re.sub('\\$|\\s|[a-zA-Z]','', s)\n",
    "            value = float(s) * 10**6\n",
    "            return value\n",
    "        elif re.match(r'\\$\\s*\\d+\\.?\\d*\\s*billi?on', s, flags=re.IGNORECASE):\n",
    "            s = re.sub('\\$|\\s|[a-zA-Z]','', s)\n",
    "            value = float(s) * 10**9\n",
    "            return value\n",
    "        elif re.match(r'\\$\\s*\\d{1,3}(?:[,\\.]\\d{3})+(?!\\s[mb]illion)', s, flags=re.IGNORECASE):\n",
    "            s = re.sub('\\$|,','', s)\n",
    "            value = float(s)\n",
    "            return value\n",
    "        else:\n",
    "            return np.nan    \n",
    "        \n",
    "    # Clean the box office column in the wiki_movies_df DataFrame.\n",
    "    wiki_movies_df['box_office'] = box_office.str.extract(f'({form_one}|{form_two})', flags=re.IGNORECASE)[0].apply(parse_dollars)\n",
    "    wiki_movies_df.drop('Box office', axis=1, inplace=True)\n",
    "    \n",
    "    # Clean the budget column in the wiki_movies_df DataFrame.\n",
    "    budget = wiki_movies_df['Budget'].dropna()\n",
    "    \n",
    "    budget = budget.map(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    \n",
    "    budget = budget.str.replace(r'\\$.*[-—–](?![a-z])', '$', regex=True)\n",
    "    \n",
    "    wiki_movies_df['budget'] = budget.str.extract(f'({form_one}|{form_two})', flags=re.IGNORECASE)[0].apply(parse_dollars)\n",
    "    \n",
    "    wiki_movies_df.drop('Budget', axis=1, inplace=True)     \n",
    "\n",
    "    # Clean the release date column in the wiki_movies_df DataFrame.\n",
    "    release_date = wiki_movies_df['Release date'].dropna().apply(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    \n",
    "    date_form_one = r'(?:January|February|March|April|May|June|July|August|September|October|November|December)\\s[123]\\d,\\s\\d{4}'\n",
    "    date_form_two = r'\\d{4}.[01]\\d.[123]\\d'\n",
    "    date_form_three = r'(?:January|February|March|April|May|June|July|August|September|October|November|December)\\s\\d{4}'\n",
    "    date_form_four = r'\\d{4}'\n",
    "    \n",
    "    wiki_movies_df['release_date'] = pd.to_datetime(release_date.str.extract(f'({date_form_one}|{date_form_two}|{date_form_three}|{date_form_four})')[0], infer_datetime_format=True)\n",
    "     \n",
    "\n",
    "    # Clean the running time column in the wiki_movies_df DataFrame.\n",
    "    running_time = wiki_movies_df['Running time'].dropna().apply(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    \n",
    "    running_time_extract = running_time.str.extract(r'(\\d+)\\s*ho?u?r?s?\\s*(\\d*)|(\\d+)\\s*m')\n",
    "    \n",
    "    running_time_extract = running_time_extract.apply(lambda col: pd.to_numeric(col, errors='coerce')).fillna(0)\n",
    "    \n",
    "    wiki_movies_df['running_time'] = running_time_extract.apply(lambda row: row[0]*60 + row[1] if row[2] == 0 else row[2], axis=1)\n",
    "    \n",
    "    wiki_movies_df.drop('Running time', axis=1, inplace=True)\n",
    "    \n",
    "     \n",
    "    # 2. Clean the Kaggle metadata.\n",
    "    kaggle_metadata = kaggle_metadata[kaggle_metadata['adult'] == 'False'].drop('adult',axis='columns')  \n",
    "    kaggle_metadata['video'] = kaggle_metadata['video'] == 'True'\n",
    "    kaggle_metadata['budget'] = kaggle_metadata['budget'].astype(int)\n",
    "    kaggle_metadata['id'] = pd.to_numeric(kaggle_metadata['id'], errors='raise')\n",
    "    kaggle_metadata['popularity'] = pd.to_numeric(kaggle_metadata['popularity'], errors='raise')\n",
    "    kaggle_metadata['release_date'] = pd.to_datetime(kaggle_metadata['release_date'])\n",
    "    ratings['timestamp'] = pd.to_datetime(ratings['timestamp'], unit='s')\n",
    "    \n",
    "    # 3. Merged the two DataFrames into the movies DataFrame.\n",
    "    movies_df = pd.merge(wiki_movies_df, kaggle_metadata, on='imdb_id', suffixes=['_wiki','_kaggle'])\n",
    "    \n",
    "    \n",
    "    # 4. Drop unnecessary columns from the merged DataFrame.\n",
    "    movies_df.drop(columns=['title_wiki','release_date_wiki','Language','Production company(s)'], inplace=True)\n",
    "\n",
    "    # 5. Add in the function to fill in the missing Kaggle data.\n",
    "    def fill_missing_kaggle_data(df, kaggle_column, wiki_column):\n",
    "        df[kaggle_column] = df.apply(\n",
    "        lambda row: row[wiki_column] if row[kaggle_column] == 0 else row[kaggle_column]\n",
    "        , axis=1)\n",
    "        df.drop(columns=wiki_column, inplace=True)\n",
    "    \n",
    "    # 6. Call the function in Step 5 with the DataFrame and columns as the arguments.\n",
    "    fill_missing_kaggle_data(movies_df, 'runtime', 'running_time')\n",
    "    fill_missing_kaggle_data(movies_df, 'budget_kaggle', 'budget_wiki')\n",
    "    fill_missing_kaggle_data(movies_df, 'revenue', 'box_office')\n",
    "\n",
    "    # 7. Filter the movies DataFrame for specific columns.\n",
    "    movies_df = movies_df[['imdb_id','id','title_kaggle','original_title','tagline','belongs_to_collection','url','imdb_link',\n",
    "                        'runtime','budget_kaggle','revenue','release_date_kaggle','popularity','vote_average','vote_count',\n",
    "                        'genres','original_language','overview','spoken_languages','Country',\n",
    "                        'production_companies','production_countries','Distributor',\n",
    "                        'Producer(s)','Director','Starring','Cinematography','Editor(s)','Writer(s)','Composer(s)','Based on'\n",
    "                        ]]\n",
    "\n",
    "    # 8. Rename the columns in the movies DataFrame.\n",
    "    movies_df.rename({'id':'kaggle_id',\n",
    "                  'title_kaggle':'title',\n",
    "                  'url':'wikipedia_url',\n",
    "                  'budget_kaggle':'budget',\n",
    "                  'release_date_kaggle':'release_date',\n",
    "                  'Country':'country',\n",
    "                  'Distributor':'distributor',\n",
    "                  'Producer(s)':'producers',\n",
    "                  'Director':'director',\n",
    "                  'Starring':'starring',\n",
    "                  'Cinematography':'cinematography',\n",
    "                  'Editor(s)':'editors',\n",
    "                  'Writer(s)':'writers',\n",
    "                  'Composer(s)':'composers',\n",
    "                  'Based on':'based_on'\n",
    "                 }, axis='columns', inplace=True)\n",
    "    \n",
    "    \n",
    "    # 9. Transform and merge the ratings DataFrame.\n",
    "    rating_counts = ratings.groupby(['movieId','rating'], as_index=False).count() \\\n",
    "                .rename({'userId':'count'}, axis=1) \\\n",
    "                .pivot(index='movieId',columns='rating', values='count')\n",
    "    \n",
    "    rating_counts.columns = ['rating_' + str(col) for col in rating_counts.columns]\n",
    "    \n",
    "    movies_with_ratings_df = pd.merge(movies_df, rating_counts, left_on='kaggle_id', right_index=True, how='left')\n",
    "    \n",
    "    movies_with_ratings_df[rating_counts.columns] = movies_with_ratings_df[rating_counts.columns].fillna(0)\n",
    "    \n",
    "    print(\"Complete up until db\")\n",
    "    \n",
    "    # Create the Database Engine - local server, the connection string will be as follows:\n",
    "    db_string = f\"postgresql://postgres:{db_password}@127.0.0.1:5432/movie_data\"\n",
    "    \n",
    "    # Create the database engine (to the PostgreSQL database)\n",
    "    engine = create_engine(db_string)\n",
    "    \n",
    "    # Import the movies_df DataFrame to a SQL database.\n",
    "    # movies_df.to_sql(name='movies', con=engine)\n",
    "    # movies_df.to_sql(name='movies', if_exists='replace',con=engine)\n",
    "    movies_df.to_sql(name='movies', if_exists='replace',con=engine)\n",
    "    # Create a variable rows_imported\n",
    "    rows_imported = 0\n",
    "    \n",
    "    # Set start_time from time.time()\n",
    "    start_time = time.time()\n",
    "    #count = 0\n",
    "    \n",
    "# create a variable for the number of rows imported\n",
    "    for data in pd.read_csv(f'{file_dir}ratings.csv', chunksize=1000000):\n",
    "        #if_exits =\"\"\n",
    "        #if count <= 0:\n",
    "            #if_exits = 'replace'\n",
    "        #else:\n",
    "            #if_exists = 'append'\n",
    "            \n",
    "        # Print out the range of rows\n",
    "        print(f'importing rows {rows_imported} to {rows_imported + len(data)}...', end='')\n",
    "\n",
    "        # print out the range of rows that are being imported\n",
    "        data.to_sql(name='ratings', con=engine, if_exists= 'append')\n",
    "\n",
    "        # Increment the numb. rows imported\n",
    "        rows_imported += len(data)\n",
    "\n",
    "        # Final print out, add elapsed time to final print out\n",
    "        #count +=1\n",
    "        print(f'Done. {time.time() - start_time} total seconds elapsed')\n",
    "    print('loop has finished')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "31bd9b48",
   "metadata": {
    "id": "31bd9b48"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Complete up until db\n",
      "importing rows 0 to 1000000...Done. 151.00808763504028 total seconds elapsed\n",
      "importing rows 1000000 to 2000000...Done. 350.7024939060211 total seconds elapsed\n",
      "importing rows 2000000 to 3000000...Done. 483.5509192943573 total seconds elapsed\n",
      "importing rows 3000000 to 4000000...Done. 571.1154744625092 total seconds elapsed\n",
      "importing rows 4000000 to 5000000...Done. 656.7197484970093 total seconds elapsed\n",
      "importing rows 5000000 to 6000000...Done. 725.3044142723083 total seconds elapsed\n",
      "importing rows 6000000 to 7000000...Done. 791.3570284843445 total seconds elapsed\n",
      "importing rows 7000000 to 8000000...Done. 850.5146491527557 total seconds elapsed\n",
      "importing rows 8000000 to 9000000...Done. 917.1847188472748 total seconds elapsed\n",
      "importing rows 9000000 to 10000000...Done. 969.0790281295776 total seconds elapsed\n",
      "importing rows 10000000 to 11000000...Done. 1021.2893569469452 total seconds elapsed\n",
      "importing rows 11000000 to 12000000...Done. 1080.6960389614105 total seconds elapsed\n",
      "importing rows 12000000 to 13000000...Done. 1140.7423293590546 total seconds elapsed\n",
      "importing rows 13000000 to 14000000...Done. 1196.0999999046326 total seconds elapsed\n",
      "importing rows 14000000 to 15000000...Done. 1254.6556541919708 total seconds elapsed\n",
      "importing rows 15000000 to 16000000...Done. 1315.1202638149261 total seconds elapsed\n",
      "importing rows 16000000 to 17000000...Done. 1403.4363305568695 total seconds elapsed\n",
      "importing rows 17000000 to 18000000...Done. 1464.2885990142822 total seconds elapsed\n",
      "importing rows 18000000 to 19000000...Done. 1521.5848333835602 total seconds elapsed\n",
      "importing rows 19000000 to 20000000...Done. 1589.322429895401 total seconds elapsed\n",
      "importing rows 20000000 to 21000000...Done. 1694.958729505539 total seconds elapsed\n",
      "importing rows 21000000 to 22000000...Done. 1779.1902923583984 total seconds elapsed\n",
      "importing rows 22000000 to 23000000...Done. 1871.1870398521423 total seconds elapsed\n",
      "importing rows 23000000 to 24000000...Done. 1986.0867466926575 total seconds elapsed\n",
      "importing rows 24000000 to 25000000...Done. 2066.5547761917114 total seconds elapsed\n",
      "importing rows 25000000 to 26000000...Done. 2129.8712282180786 total seconds elapsed\n",
      "importing rows 26000000 to 26024289...Done. 2131.204159259796 total seconds elapsed\n",
      "loop has finished\n"
     ]
    }
   ],
   "source": [
    "#Call the function\n",
    "extract_transform_load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2d5989a8",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "colab": {
   "name": "ETL_create_database.ipynb",
   "provenance": []
  },
  "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.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}