ETL_clean_kaggle_data.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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",
    "\n",
    "# from config import db_password\n",
    "\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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",
    "    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",
    "    return movie"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 1 Add the function that takes in three arguments;\n",
    "\n",
    "def three_arguments_func():\n",
    "    # Read in the movies_metadata and MovieLens ratings CSV files as Pandas DataFrames.\n",
    "    movies_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",
    "    # Open and read the Wikipedia data JSON file.\n",
    "    f'{file_dir}wikipedia-movies.json'\n",
    "    with open(f'{file_dir}/wikipedia-movies.json', mode='r') as file:\n",
    "        wiki_movies_raw = json.load(file)\n",
    "        wiki_movies_df = pd.DataFrame(wiki_movies_raw)    \n",
    "    \n",
    "    # Write a list comprehension to filter out TV shows.\n",
    "    wiki_tv = [tvshows for tvshows in wiki_movies_raw \n",
    "        if 'Television series' in tvshows]     \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_raw]     \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: \n",
    "        print(\"No link avialable\") \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.map(lambda x: type(x) != str)]     \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",
    "    matches_form_one = budget.str.contains(form_one, flags=re.IGNORECASE)\n",
    "    matches_form_two = budget.str.contains(form_two, flags=re.IGNORECASE)\n",
    "    budget[~matches_form_one & ~matches_form_two]\n",
    "    \n",
    "    budget = budget.str.replace(r'\\[\\d+\\]\\s*', '')\n",
    "    budget[~matches_form_one & ~matches_form_two]\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",
    "    release_date.str.extract(f'({date_form_one}|{date_form_two}|{date_form_three}|{date_form_four})', flags=re.IGNORECASE)\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",
    "    movies_metadata[~movies_metadata['adult'].isin(['True','False'])]\n",
    "    \n",
    "    movies_metadata = movies_metadata[movies_metadata['adult'] == 'False'].drop('adult',axis='columns')\n",
    "    \n",
    "    movies_metadata ['video'] == 'True'\n",
    "    \n",
    "    movies_metadata['video'] = movies_metadata['video'] == 'True'\n",
    "    \n",
    "    movies_metadata['budget'] = movies_metadata['budget'].astype(int)\n",
    "    movies_metadata['id'] = pd.to_numeric(movies_metadata['id'], errors='raise')\n",
    "    movies_metadata['popularity'] = pd.to_numeric(movies_metadata['popularity'], errors='raise')\n",
    "    \n",
    "    movies_metadata['release_date'] = pd.to_datetime(movies_metadata['release_date'])\n",
    "    \n",
    "    pd.to_datetime(ratings['timestamp'], unit='s')\n",
    "    \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, movies_metadata, on='imdb_id', suffixes=['_wiki','_kaggle'])\n",
    "\n",
    "    # 4. Drop unnecessary columns from the merged DataFrame.\n",
    "\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",
    "    movies_df\n",
    "\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",
    "    # 8. Rename the columns in the movies DataFrame.\n",
    "    column_names = ['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",
    "    movies_df = movies_df.reindex(columns=column_names)\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",
    "    \n",
    "    # 9. Transform and merge the ratings DataFrame.\n",
    "    rating_counts = ratings.groupby(['movieId','rating'], as_index=False).count()\n",
    "    \n",
    "    rating_counts = ratings.groupby(['movieId','rating'], as_index=False).count() \\\n",
    "                .rename({'userId':'count'}, axis=1)\n",
    "    \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",
    "    return wiki_movies_df, movies_with_ratings_df, movies_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 10. Create the path to your file directory and variables for the three files.\n",
    "file_dir = \"C://Users/jacqu/Desktop/Class Folder/Movies-ETL/\"\n",
    "# The Wikipedia data\n",
    "wiki_file = f'{file_dir}/wikipedia.movies.json'\n",
    "# The Kaggle metadata\n",
    "movies_metadata_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": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# 11. Set the three variables equal to the function created in D1.\n",
    "wiki_file, movies_metadata_file, ratings_file = three_arguments_func()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 12. Set the DataFrames from the return statement equal to the file names in Step 11. \n",
    "wiki_movies_df = wiki_file\n",
    "movies_with_ratings_df = movies_metadata_file\n",
    "movies_df = ratings_file\n",
    "\n",
    "\n",
    "# 12. Set the DataFrames from the return statement equal to the file names in Step 11. \n",
    "wiki_movies_df = wiki_file\n",
    "movies_with_ratings_df = movies_metadata_file\n",
    "movies_df = ratings_file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 13. Check the wiki_movies_df DataFrame. \n",
    "wiki_movies_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 14. Check the movies_with_ratings_df DataFrame.\n",
    "movies_with_ratings_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 15. Check the movies_df DataFrame. \n",
    "movies_df.head()"
   ]
  },
  {
   "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": "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": 2
}