ETL_create_database.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import re\n",
    "from sqlalchemy import create_engine\n",
    "import psycopg2\n",
    "from config import db_password\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "#  Add the clean movie function that takes in the argument, \"movie\".\n",
    "def clean_movie(movie):\n",
    "    movie = dict(movie) #create a non-destructive copy\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",
    "    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,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 1 Add the function that takes in three arguments;\n",
    "# Wikipedia data, Kaggle metadata, and MovieLens rating data (from Kaggle)\n",
    "\n",
    "def extract_transform_load():\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')\n",
    "    ratings = pd.read_csv(f'{file_dir}ratings.csv')\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",
    "    # Write a list comprehension to filter out TV shows.\n",
    "    wiki_movies = [movie for movie in wiki_movies_raw\n",
    "               if ('Director' in movie or 'Directed by' in movie)\n",
    "                   and 'imdb_link' in movie\n",
    "                   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 error:\n",
    "        print(error)\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 s is not a string, return NaN\n",
    "        if type(s) != str:\n",
    "            return np.nan\n",
    "\n",
    "        # if input is of the form $###.# million\n",
    "        if re.match(r'\\$\\s*\\d+\\.?\\d*\\s*milli?on', s, flags=re.IGNORECASE):\n",
    "\n",
    "            # remove dollar sign and \" million\"\n",
    "            s = re.sub('\\$|\\s|[a-zA-Z]','', s)\n",
    "\n",
    "            # convert to float and multiply by a million\n",
    "            value = float(s) * 10**6\n",
    "\n",
    "            # return value\n",
    "            return value\n",
    "\n",
    "        # if input is of the form $###.# billion\n",
    "        elif re.match(r'\\$\\s*\\d+\\.?\\d*\\s*billi?on', s, flags=re.IGNORECASE):\n",
    "\n",
    "            # remove dollar sign and \" billion\"\n",
    "            s = re.sub('\\$|\\s|[a-zA-Z]','', s)\n",
    "\n",
    "            # convert to float and multiply by a billion\n",
    "            value = float(s) * 10**9\n",
    "\n",
    "            # return value\n",
    "            return value\n",
    "\n",
    "        # if input is of the form $###,###,###\n",
    "        elif re.match(r'\\$\\s*\\d{1,3}(?:[,\\.]\\d{3})+(?!\\s[mb]illion)', s, flags=re.IGNORECASE):\n",
    "\n",
    "            # remove dollar sign and commas\n",
    "            s = re.sub('\\$|,','', s)\n",
    "\n",
    "            # convert to float\n",
    "            value = float(s)\n",
    "\n",
    "            # return value\n",
    "            return value\n",
    "\n",
    "        # otherwise, return NaN\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",
    "    budget = budget.map(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    budget = budget.str.replace(r'\\$.*[-—–](?![a-z])', '$', regex=True)\n",
    "    wiki_movies_df['budget'] = budget.str.extract(f'({form_one}|{form_two})', flags=re.IGNORECASE)[0].apply(parse_dollars)\n",
    "    \n",
    "    # Clean the release date column in the wiki_movies_df DataFrame.\n",
    "\n",
    "    release_date = wiki_movies_df['Release date'].dropna().apply(lambda x: ' '.join(x) if type(x) == list else x)\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",
    "    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",
    "    # 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",
    "    running_time_extract = running_time.str.extract(r'(\\d+)\\s*ho?u?r?s?\\s*(\\d*)|(\\d+)\\s*m')\n",
    "    running_time_extract = running_time_extract.apply(lambda col: pd.to_numeric(col, errors='coerce')).fillna(0)\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",
    "    wiki_movies_df.drop('Running time', axis=1, inplace=True)\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",
    "\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",
    "    movies_df = movies_df.drop(movies_df[(movies_df['release_date_wiki'] > '1996-01-01') & (movies_df['release_date_kaggle'] < '1965-01-01')].index)\n",
    "    \n",
    "    # 4. Drop unnecessary columns from the merged DataFrame.\n",
    "    movies_df.drop(columns=['title_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",
    "    for col in movies_df.columns:\n",
    "        lists_to_tuples = lambda x: tuple(x) if type(x) == list else x\n",
    "        value_counts = movies_df[col].apply(lists_to_tuples).value_counts(dropna=False)\n",
    "        num_values = len(value_counts)\n",
    "        if num_values == 1:\n",
    "            print(col)\n",
    "    \n",
    "    movies_df['video'].value_counts(dropna=False)\n",
    "    \n",
    "    # 8. Rename the columns in the movies DataFrame.\n",
    "    movies_df = movies_df.loc[:, ['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",
    "    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",
    "    # 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",
    "    rating_counts.columns = ['rating_' + str(col) for col in rating_counts.columns]\n",
    "    movies_with_ratings_df = pd.merge(movies_df, rating_counts, left_on='kaggle_id', right_index=True, how='left')\n",
    "    movies_with_ratings_df[rating_counts.columns] = movies_with_ratings_df[rating_counts.columns].fillna(0)\n",
    "    \n",
    "    \n",
    "    db_string = f\"postgresql://postgres:{db_password}@127.0.0.1:5432/movie_data\"\n",
    "    engine = create_engine(db_string)\n",
    "    movies_df.to_sql(name='movies', con=engine, if_exists='replace')\n",
    "\n",
    "    \n",
    "    rows_imported = 0\n",
    "    # get the start_time from time.time()\n",
    "    start_time = time.time()\n",
    "    for data in pd.read_csv(f'{file_dir}/ratings.csv', chunksize=1000000):\n",
    "        print(f'importing rows {rows_imported} to {rows_imported + len(data)}...', end='')\n",
    "        data.to_sql(name='ratings', con=engine, if_exists='append')\n",
    "        rows_imported += len(data)\n",
    "\n",
    "        # add elapsed time to final print out\n",
    "        print(f'Done. {time.time() - start_time} total seconds elapsed')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 10. Create the path to your file directory and variables for the three files.\n",
    "file_dir = '/Users/michaelbinger/Documents/Bootcamp/Module_8_ETL/Challenge/'\n",
    "# The Wikipedia data\n",
    "wiki_file = f'{file_dir}wikipedia_movies.json'\n",
    "# The Kaggle metadata\n",
    "kaggle_file = f'{file_dir}movies_metadata.csv'\n",
    "# The MovieLens rating data.\n",
    "ratings_file = f'{file_dir}ratings.csv'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/var/folders/2d/w0_y9zt547v2y0371_wclbcm0000gn/T/ipykernel_42539/1057435404.py:2: DtypeWarning: Columns (10) have mixed types.Specify dtype option on import or set low_memory=False.\n",
      "  extract_transform_load()\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "video\n",
      "importing rows 0 to 1000000...Done. 42.001384019851685 total seconds elapsed\n",
      "importing rows 1000000 to 2000000...Done. 84.22941708564758 total seconds elapsed\n",
      "importing rows 2000000 to 3000000...Done. 123.77263617515564 total seconds elapsed\n",
      "importing rows 3000000 to 4000000...Done. 158.13901615142822 total seconds elapsed\n",
      "importing rows 4000000 to 5000000...Done. 194.8413541316986 total seconds elapsed\n",
      "importing rows 5000000 to 6000000...Done. 230.971097946167 total seconds elapsed\n",
      "importing rows 6000000 to 7000000...Done. 266.7155010700226 total seconds elapsed\n",
      "importing rows 7000000 to 8000000...Done. 303.14479303359985 total seconds elapsed\n",
      "importing rows 8000000 to 9000000...Done. 338.8964509963989 total seconds elapsed\n",
      "importing rows 9000000 to 10000000...Done. 370.56506419181824 total seconds elapsed\n",
      "importing rows 10000000 to 11000000...Done. 406.222403049469 total seconds elapsed\n",
      "importing rows 11000000 to 12000000...Done. 439.6307511329651 total seconds elapsed\n",
      "importing rows 12000000 to 13000000...Done. 476.4424858093262 total seconds elapsed\n",
      "importing rows 13000000 to 14000000...Done. 515.6177382469177 total seconds elapsed\n",
      "importing rows 14000000 to 15000000...Done. 557.1864950656891 total seconds elapsed\n",
      "importing rows 15000000 to 16000000...Done. 595.0570349693298 total seconds elapsed\n",
      "importing rows 16000000 to 17000000...Done. 630.6782219409943 total seconds elapsed\n",
      "importing rows 17000000 to 18000000...Done. 668.2227680683136 total seconds elapsed\n",
      "importing rows 18000000 to 19000000...Done. 703.4589569568634 total seconds elapsed\n",
      "importing rows 19000000 to 20000000...Done. 737.4581549167633 total seconds elapsed\n",
      "importing rows 20000000 to 21000000...Done. 774.0989751815796 total seconds elapsed\n",
      "importing rows 21000000 to 22000000...Done. 807.4496929645538 total seconds elapsed\n",
      "importing rows 22000000 to 23000000...Done. 842.3276610374451 total seconds elapsed\n",
      "importing rows 23000000 to 24000000...Done. 878.9808259010315 total seconds elapsed\n",
      "importing rows 24000000 to 25000000...Done. 915.2919709682465 total seconds elapsed\n",
      "importing rows 25000000 to 26000000...Done. 952.2793262004852 total seconds elapsed\n",
      "importing rows 26000000 to 26024289...Done. 953.1705770492554 total seconds elapsed\n"
     ]
    }
   ],
   "source": [
    "# 11. Set the three variables equal to the function created in D1.\n",
    "extract_transform_load()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PythonData",
   "language": "python",
   "name": "pythondata"
  },
  "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.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}