ETL_Challenge.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import dependencies / modules\n",
    "\n",
    "import time\n",
    "import re\n",
    "from sqlalchemy import create_engine\n",
    "import json\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import psycopg2\n",
    "from config import db_password"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define variable for database connection\n",
    "protocol = \"postgres\"\n",
    "user = \"postgres\"\n",
    "location = \"localhost\"\n",
    "port = \"5432\"\n",
    "db = \"movie_data\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "connection = f'{protocol}://{user}:{db_password}@{location}:{port}/{db}'\n",
    "engine = create_engine(connection)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def clean_movie(movie): # Create a function for cleaning. This is unique to wiki_file only. \n",
    "    \n",
    "    movie = dict(movie) #create a non-destructive copy of movie\n",
    "  \n",
    "    alt_titles = {}     #Create an empty list for alt_title \n",
    "                        # combine alternate titles into one list\n",
    "        \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",
    "        \n",
    "        if key in movie:\n",
    "            alt_titles[key] = movie[key]\n",
    "            movie.pop(key)\n",
    "   \n",
    "    if len(alt_titles) > 0:    # Use the length of alt_titles to check or validate value\n",
    "        movie['alt_titles'] = alt_titles  # Convert to alt_titles i.e combine alternate title into one list\n",
    "\n",
    "    # merge column names\n",
    "    def change_column_name(old_name, new_name):  # Create a function change_column_name old and new \n",
    "        \n",
    "        if old_name in movie:\n",
    "            movie[new_name] = movie.pop(old_name)   # Use pop() method to consolidate columns with\n",
    "                                                    # same data into one or change the name of a key\n",
    "   \n",
    "    change_column_name('Adaptation by', 'Writer(s)') # Go through and decide if there’s a better name for it.\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\n",
    "\n",
    "\n",
    "def etl_pipe_func(wiki_file, kaggle_file, ratings_file):\n",
    "        \n",
    "    # Open/Load and read the first file, a .json raw file, using \"with open()\" \n",
    "    with open(f'{file_dir}/wikipedia.movies.json', mode='r') as file:\n",
    "        wiki_movies_raw = json.load(file)\n",
    "    \n",
    "    # Read the second file, a csv file, using Pandas pd.read_csv() function    \n",
    "    kaggle_metadata = pd.read_csv(f'{file_dir}/movies_metadata.csv')\n",
    "\n",
    "    # Read the third file, a csv file, using Pandas pd.read_csv() function\n",
    "    ratings = pd.read_csv(f'{file_dir}/ratings.csv')\n",
    "\n",
    "    \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",
    "    clean_movies = [clean_movie(movie) for movie in wiki_movies]\n",
    "    \n",
    "    wiki_movies_df = pd.DataFrame(clean_movies)\n",
    "    \n",
    "    # This is the beginig of cleaning the first file\n",
    "    # Use imdb_id to check as we clean out duplicates and drop NAs. Extact \"imdb_id\" from imdb_link\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",
    "    except Exception as exception:\n",
    "        print('exception')\n",
    "        \n",
    "    # Apply list comprehension    \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",
    "    # Drop Nas in wiki_movies Box Office\n",
    "    box_office = wiki_movies_df['Box office'].dropna() \n",
    "    box_office = box_office.apply(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    \n",
    "    # Apply regular expression\n",
    "    form_one = r'\\$\\d+\\.?\\d*\\s*[mb]illion' # Create a variable for the first regular expression\n",
    "                                       # To count up how many box office values match our first form\n",
    "                                       # Use str.contain() method. Set the flag to ignore case\n",
    "    form_two = r'\\$\\d{1,3}(?:,\\d{3})+'    \n",
    "    \n",
    "    # Parse the Box Office Data to Numeric values\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",
    "    \n",
    "    # First we need to extract the values from Box Office using .str.extract. Then, apply\n",
    "    # parse_dollars to the first column in the DataFrame returned by .str.extract\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",
    "    # Create a budget variable \n",
    "    budget = wiki_movies_df['Budget'].dropna()\n",
    "\n",
    "    # Convert any lists to strings:\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",
    "    \n",
    "    # Parse Budget using the same code as in Box Office. Replace box_office with budget\n",
    "    wiki_movies_df['budget'] = budget.str.extract(f'({form_one}|{form_two})', flags=re.IGNORECASE)[0].apply(parse_dollars)\n",
    "    wiki_movies_df.drop('budget', axis=1, inplace=True)\n",
    "    \n",
    "    release_date = wiki_movies_df['Release date'].dropna().apply(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    \n",
    "    # Convert dates to forms\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",
    "    # Dropna from Wiki_movies\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",
    "    kaggle_metadata = kaggle_metadata[kaggle_metadata['adult'] == 'False'].drop('adult',axis='columns')\n",
    "    kaggle_metadata['video'] = kaggle_metadata['video'] == 'True'\n",
    "    \n",
    "    # Convert Kaggle to Numeric values\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",
    "    \n",
    "    # Merge Wikipedia data and Kaggle data after being cleaned up and in tabular formats\n",
    "    # The main purpose is to avoid duplicates. We also drop some columns that are redundant.\n",
    "    movies_df = pd.merge(wiki_movies_df, kaggle_metadata, on='imdb_id', suffixes=['_wiki','_kaggle'])\n",
    "    movies_df.drop(columns=['title_wiki','release_date_wiki','Language','Production company(s)'], inplace=True)\n",
    "\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",
    "    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",
    "    # Reorder the columns to make the information easier to read and understand\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",
    "    # For consistency we rename the columns in 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",
    "    rating_counts = ratings.groupby(['movieId','rating'], as_index=False).count() \\\n",
    "                .rename({'userId':'count'}, axis=1)\n",
    "    rating_counts.columns = ['rating_' + str(col) for col in rating_counts.columns]\n",
    "    \n",
    "    # Merge the data between files\n",
    "    movies_with_ratings_df = pd.merge(movies_df, rating_counts, left_on='kaggle_id', right_index=True, how='left')\n",
    "    \n",
    "    # Fill in zeros for missing values\n",
    "    movies_with_ratings_df[rating_counts.columns] = movies_with_ratings_df[rating_counts.columns].fillna(0)\n",
    "    \n",
    "    # Define variables for database connection\n",
    "    protocol = \"postgres\"\n",
    "    user = \"postgres2020\"\n",
    "    location = \"localhost\"\n",
    "    port = \"5432\"\n",
    "    db = \"movie_data\"\n",
    "    \n",
    "    # Provide connection string to Postgres db\n",
    "    connection = f'{protocol}://{user}:{db_password}@{location}:{port}/{db}'\n",
    "    \n",
    "    # Create database engine\n",
    "    engine = create_engine(connection)\n",
    "    \n",
    "    # Import Movie data to the sql database\n",
    "    movies_df.to_sql(name='movies', con=engine)\n",
    "    \n",
    "    # create a variable for the number of rows imported\n",
    "    rows_imported = 0\n",
    "    for data in pd.read_csv(f'{file_dir}ratings.csv', chunksize=1000000):\n",
    "    \n",
    "    # Print out the range of rows that are being imported\n",
    "        print(f'importing rows {rows_imported} to {rows_imported + len(data)}...', end='')\n",
    "    \n",
    "    data.to_sql(name='ratings', con=engine, if_exists='append')\n",
    "    \n",
    "    # Increment the number of rows imported by the size of 'data'\n",
    "    rows_imported += len(data)\n",
    "    \n",
    "    # Print that the rows have finished importing\n",
    "    print('Done importing...........')\n",
    "    \n",
    "    # This is the working directory path for all resource files\n",
    "    file_dir = 'C:/Users/Franc/Berkeley/class_folder/Movies_ETL/Resources'\n",
    "    \n",
    "    wiki_file = f'{file_dir}/wikipedia.movies.json'\n",
    "    kaggle_file = f'{file_dir}/movies_metadata.csv'\n",
    "    ratings_file = f'{file_dir}/ratings_csv'\n",
    "    \n",
    "    # Call the main function\n",
    "    \n",
    "    etl_pipe_func(wiki_file, kaggle_file, ratings_file)\n",
    "    \n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "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.7.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}