prestudy_of_dataframes.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "ca795588-0157-4d17-a0e8-50733dde384a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# After we examine this file, we will use these functions in transformation.py file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "854078ea-648a-44f2-94d8-7a70be0298a0",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession, functions as F\n",
    "import findspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "53b180c5-158c-47e8-b5b0-b43bbe1da271",
   "metadata": {},
   "outputs": [],
   "source": [
    "findspark.init(\"/opt/manual/spark\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "d04cba00-a149-4419-9f92-51bda399036b",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "b320d880-6c53-402c-8afd-df747a2fe6ee",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: An illegal reflective access operation has occurred\n",
      "WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/opt/manual/spark-3.1.1-bin-hadoop3.2/jars/spark-unsafe_2.12-3.1.1.jar) to constructor java.nio.DirectByteBuffer(long,int)\n",
      "WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform\n",
      "WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations\n",
      "WARNING: All illegal access operations will be denied in a future release\n",
      "2022-11-27 17:07:25,568 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "# Create spark session.\n",
    "spark = (SparkSession.builder\n",
    "         .appName(\"prestudy\")\n",
    "         .config(\"spark.hadoop.fs.s3a.access.key\", \"root\")\n",
    "         .config(\"spark.hadoop.fs.s3a.secret.key\", \"root12345\")\n",
    "         .config(\"spark.hadoop.fs.s3a.path.style.access\", \"true\")\n",
    "         .config(\"spark.hadoop.fs.s3a.impl\", \"org.apache.hadoop.fs.s3a.S3AFileSystem\")\n",
    "         .config(\"spark.hadoop.fs.s3a.endpoint\", \"http://localhost:9000\")\n",
    "         .getOrCreate())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1217d000-91e5-43cd-8318-36af39b922e8",
   "metadata": {
    "tags": []
   },
   "source": [
    "### CREDITS - RAW"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "28b39936-d1d1-4cc9-a891-bb636b51ab32",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Read credits raw data.\n",
    "df_credits = spark.read.format(\"parquet\").load(\"s3a://tmdb-bronze/credits/credits_part_20221124-130321.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "33003a33-0551-4f55-9536-3fbc9b3d0711",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 1:>                                                          (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------------+--------------------+--------------------+--------------------+\n",
      "|movie_id|               title|                cast|                crew|          event_time|\n",
      "+--------+--------------------+--------------------+--------------------+--------------------+\n",
      "|   19995|              Avatar|[{\"cast_id\": 242,...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|     285|Pirates of the Ca...|[{\"cast_id\": 4, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|  206647|             Spectre|[{\"cast_id\": 1, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   49026|The Dark Knight R...|[{\"cast_id\": 2, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   49529|         John Carter|[{\"cast_id\": 5, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|     559|        Spider-Man 3|[{\"cast_id\": 30, ...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   38757|             Tangled|[{\"cast_id\": 34, ...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   99861|Avengers: Age of ...|[{\"cast_id\": 76, ...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|     767|Harry Potter and ...|[{\"cast_id\": 3, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|  209112|Batman v Superman...|[{\"cast_id\": 18, ...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|    1452|    Superman Returns|[{\"cast_id\": 3, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   10764|   Quantum of Solace|[{\"cast_id\": 1, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|      58|Pirates of the Ca...|[{\"cast_id\": 37, ...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   57201|     The Lone Ranger|[{\"cast_id\": 4, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   49521|        Man of Steel|[{\"cast_id\": 2, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|    2454|The Chronicles of...|[{\"cast_id\": 1, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   24428|        The Avengers|[{\"cast_id\": 46, ...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|    1865|Pirates of the Ca...|[{\"cast_id\": 15, ...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|   41154|      Men in Black 3|[{\"cast_id\": 4, \"...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "|  122917|The Hobbit: The B...|[{\"cast_id\": 10, ...|[{\"credit_id\": \"5...|2022-11-24 16:03:...|\n",
      "+--------+--------------------+--------------------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df_credits.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "192d3f31-2ab1-4b53-9753-5f608495d7c8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movie_id: long (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- cast: string (nullable = true)\n",
      " |-- crew: string (nullable = true)\n",
      " |-- event_time: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_credits.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4fb204f7-572c-4f67-ad83-6b7575d8dc17",
   "metadata": {
    "tags": []
   },
   "source": [
    "### MOVIES - RAW"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "34918794-7e5e-4961-8cec-1309b963636a",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read movies raw data.\n",
    "df_movies = spark.read.format(\"parquet\").load(\"s3a://tmdb-bronze/movies/movies_part_20221124-130454.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "eccf1b62-a6ec-43cd-b3a6-f302f8a07ef2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------------------+------+--------------------+-----------------+--------------------+--------------------+----------+--------------------+--------------------+------------+----------+-------+--------------------+--------+--------------------+--------------------+------------+----------+--------------------+\n",
      "|   budget|              genres|            homepage|    id|            keywords|original_language|      original_title|            overview|popularity|production_companies|production_countries|release_date|   revenue|runtime|    spoken_languages|  status|             tagline|               title|vote_average|vote_count|          event_time|\n",
      "+---------+--------------------+--------------------+------+--------------------+-----------------+--------------------+--------------------+----------+--------------------+--------------------+------------+----------+-------+--------------------+--------+--------------------+--------------------+------------+----------+--------------------+\n",
      "|237000000|[{\"id\": 28, \"name...|http://www.avatar...| 19995|[{\"id\": 1463, \"na...|               en|              Avatar|In the 22nd centu...|150.437577|[{\"name\": \"Ingeni...|[{\"iso_3166_1\": \"...|  2009-12-10|2787965087|  162.0|[{\"iso_639_1\": \"e...|Released|Enter the World o...|              Avatar|         7.2|     11800|2022-11-24 16:04:...|\n",
      "|300000000|[{\"id\": 12, \"name...|http://disney.go....|   285|[{\"id\": 270, \"nam...|               en|Pirates of the Ca...|Captain Barbossa,...|139.082615|[{\"name\": \"Walt D...|[{\"iso_3166_1\": \"...|  2007-05-19| 961000000|  169.0|[{\"iso_639_1\": \"e...|Released|At the end of the...|Pirates of the Ca...|         6.9|      4500|2022-11-24 16:04:...|\n",
      "|245000000|[{\"id\": 28, \"name...|http://www.sonypi...|206647|[{\"id\": 470, \"nam...|               en|             Spectre|A cryptic message...|107.376788|[{\"name\": \"Columb...|[{\"iso_3166_1\": \"...|  2015-10-26| 880674609|  148.0|[{\"iso_639_1\": \"f...|Released|A Plan No One Esc...|             Spectre|         6.3|      4466|2022-11-24 16:04:...|\n",
      "|250000000|[{\"id\": 28, \"name...|http://www.thedar...| 49026|[{\"id\": 849, \"nam...|               en|The Dark Knight R...|Following the dea...| 112.31295|[{\"name\": \"Legend...|[{\"iso_3166_1\": \"...|  2012-07-16|1084939099|  165.0|[{\"iso_639_1\": \"e...|Released|     The Legend Ends|The Dark Knight R...|         7.6|      9106|2022-11-24 16:04:...|\n",
      "|260000000|[{\"id\": 28, \"name...|http://movies.dis...| 49529|[{\"id\": 818, \"nam...|               en|         John Carter|John Carter is a ...| 43.926995|[{\"name\": \"Walt D...|[{\"iso_3166_1\": \"...|  2012-03-07| 284139100|  132.0|[{\"iso_639_1\": \"e...|Released|Lost in our world...|         John Carter|         6.1|      2124|2022-11-24 16:04:...|\n",
      "+---------+--------------------+--------------------+------+--------------------+-----------------+--------------------+--------------------+----------+--------------------+--------------------+------------+----------+-------+--------------------+--------+--------------------+--------------------+------------+----------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_movies.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "cd94f33b-643f-4c35-8da7-4c027b6477af",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- budget: long (nullable = true)\n",
      " |-- genres: string (nullable = true)\n",
      " |-- homepage: string (nullable = true)\n",
      " |-- id: long (nullable = true)\n",
      " |-- keywords: string (nullable = true)\n",
      " |-- original_language: string (nullable = true)\n",
      " |-- original_title: string (nullable = true)\n",
      " |-- overview: string (nullable = true)\n",
      " |-- popularity: double (nullable = true)\n",
      " |-- production_companies: string (nullable = true)\n",
      " |-- production_countries: string (nullable = true)\n",
      " |-- release_date: string (nullable = true)\n",
      " |-- revenue: long (nullable = true)\n",
      " |-- runtime: double (nullable = true)\n",
      " |-- spoken_languages: string (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- tagline: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- vote_average: double (nullable = true)\n",
      " |-- vote_count: long (nullable = true)\n",
      " |-- event_time: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_movies.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "19d1b9db-0906-440e-8376-5272e7e33445",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### CAST & CREW"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "201b24d0-5eda-4027-95b3-dceaf38c6b2e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create appropriate schema for cast column to apply from_json function.\n",
    "schema_cast = (\n",
    "    ArrayType(\n",
    "        StructType([\n",
    "            StructField(\"cast_id\", IntegerType()),\n",
    "            StructField(\"character\", StringType()),\n",
    "            StructField(\"credit_id\", StringType()),\n",
    "            StructField(\"gender\", IntegerType()),\n",
    "            StructField(\"id\", IntegerType()),\n",
    "            StructField(\"name\", StringType()),\n",
    "            StructField(\"order\", IntegerType())\n",
    "                  ])\n",
    "            )\n",
    "         )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "4b0a3536-81ea-42f0-b2cd-ce8ccba9f062",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create appropriate schema for crew column to apply from_json function.\n",
    "schema_crew = (ArrayType(\n",
    "        StructType([\n",
    "            StructField(\"credit_id\", StringType()),\n",
    "            StructField(\"department\", StringType()),\n",
    "            StructField(\"gender\", IntegerType()),\n",
    "            StructField(\"id\", IntegerType()),\n",
    "            StructField(\"job\", StringType()),\n",
    "            StructField(\"name\", StringType())\n",
    "                  ])\n",
    "            )\n",
    "         )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "33b2e79c-b5ea-4d48-863c-1b0d641b71e0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create new credits datafame using from_json function to organize nested data.\n",
    "df_credits_nested= df_credits.withColumn(\"cast\", F.from_json(df_credits.cast, schema_cast)).withColumn(\"crew\", F.from_json(df_credits.crew, schema_crew))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "067cc449-d9bc-411a-9591-c80cebe42c5e",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movie_id: long (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- cast: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- cast_id: integer (nullable = true)\n",
      " |    |    |-- character: string (nullable = true)\n",
      " |    |    |-- credit_id: string (nullable = true)\n",
      " |    |    |-- gender: integer (nullable = true)\n",
      " |    |    |-- id: integer (nullable = true)\n",
      " |    |    |-- name: string (nullable = true)\n",
      " |    |    |-- order: integer (nullable = true)\n",
      " |-- crew: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- credit_id: string (nullable = true)\n",
      " |    |    |-- department: string (nullable = true)\n",
      " |    |    |-- gender: integer (nullable = true)\n",
      " |    |    |-- id: integer (nullable = true)\n",
      " |    |    |-- job: string (nullable = true)\n",
      " |    |    |-- name: string (nullable = true)\n",
      " |-- event_time: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_credits_nested.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "3cf5d72d-ddcc-4796-932d-d5b74281b2a0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Select necessary column and apply explode_outer function to create new raws for each element in the array. If the array (or map) is null or empty, explode_outer returns null. \n",
    "cast_df = df_credits_nested.select(\"movie_id\", \"title\", F.explode_outer(\"cast\")).select(\"movie_id\",\"title\", \"col.*\").drop(\"order\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "95412286-0889-40dc-a666-0f7a1db018a0",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+-------+--------------------+--------------------+------+-------+-------------------+\n",
      "|movie_id| title|cast_id|           character|           credit_id|gender|     id|               name|\n",
      "+--------+------+-------+--------------------+--------------------+------+-------+-------------------+\n",
      "|   19995|Avatar|    242|          Jake Sully|5602a8a7c3a368553...|     2|  65731|    Sam Worthington|\n",
      "|   19995|Avatar|      3|             Neytiri|52fe48009251416c7...|     1|   8691|        Zoe Saldana|\n",
      "|   19995|Avatar|     25| Dr. Grace Augustine|52fe48009251416c7...|     1|  10205|   Sigourney Weaver|\n",
      "|   19995|Avatar|      4|       Col. Quaritch|52fe48009251416c7...|     2|  32747|       Stephen Lang|\n",
      "|   19995|Avatar|      5|        Trudy Chacon|52fe48009251416c7...|     1|  17647| Michelle Rodriguez|\n",
      "|   19995|Avatar|      8|           Selfridge|52fe48009251416c7...|     2|   1771|    Giovanni Ribisi|\n",
      "|   19995|Avatar|      7|       Norm Spellman|52fe48009251416c7...|     2|  59231|   Joel David Moore|\n",
      "|   19995|Avatar|      9|                Moat|52fe48009251416c7...|     1|  30485|        CCH Pounder|\n",
      "|   19995|Avatar|     11|             Eytukan|52fe48009251416c7...|     2|  15853|          Wes Studi|\n",
      "|   19995|Avatar|     10|             Tsu'Tey|52fe48009251416c7...|     2|  10964|         Laz Alonso|\n",
      "|   19995|Avatar|     12|       Dr. Max Patel|52fe48009251416c7...|     2|  95697|         Dileep Rao|\n",
      "|   19995|Avatar|     13|      Lyle Wainfleet|52fe48009251416c7...|     2|  98215|        Matt Gerald|\n",
      "|   19995|Avatar|     32|        Private Fike|52fe48009251416c7...|     2| 154153| Sean Anthony Moran|\n",
      "|   19995|Avatar|     33| Cryo Vault Med Tech|52fe48009251416c7...|     2| 397312|        Jason Whyte|\n",
      "|   19995|Avatar|     34|Venture Star Crew...|52fe48009251416c7...|     2|  42317|     Scott Lawrence|\n",
      "|   19995|Avatar|     35|     Lock Up Trooper|52fe48009251416c7...|     2| 986734|      Kelly Kilgour|\n",
      "|   19995|Avatar|     36|       Shuttle Pilot|52fe48009251416c7...|     0|1207227| James Patrick Pitt|\n",
      "|   19995|Avatar|     37|    Shuttle Co-Pilot|52fe48009251416c7...|     0|1180936|Sean Patrick Murphy|\n",
      "|   19995|Avatar|     38|  Shuttle Crew Chief|52fe48009251416c7...|     2|1019578|       Peter Dillon|\n",
      "|   19995|Avatar|     39|Tractor Operator ...|52fe48009251416c7...|     0|  91443|       Kevin Dorman|\n",
      "+--------+------+-------+--------------------+--------------------+------+-------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "cast_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "69fb5948-fa6e-4dda-b759-2ab1a73ae413",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Select necessary column and apply explode_outer function to create new raws for each element in the array. If the array (or map) is null or empty, explode_outer returns null. \n",
    "crew_df = df_credits_nested.select(\"movie_id\", \"title\", F.explode_outer(\"crew\")).select(\"movie_id\", \"title\", \"col.*\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "eadc2c4c-5ed2-43fb-968e-b4e078b149b7",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------+--------------------+----------+------+----+--------------------+-----------------+\n",
      "|movie_id| title|           credit_id|department|gender|  id|                 job|             name|\n",
      "+--------+------+--------------------+----------+------+----+--------------------+-----------------+\n",
      "|   19995|Avatar|52fe48009251416c7...|   Editing|     0|1721|              Editor|Stephen E. Rivkin|\n",
      "|   19995|Avatar|539c47ecc3a36810e...|       Art|     2| 496|   Production Design|      Rick Carter|\n",
      "|   19995|Avatar|54491c89c3a3680fb...|     Sound|     0| 900|      Sound Designer|Christopher Boyes|\n",
      "|   19995|Avatar|54491cb70e0a26748...|     Sound|     0| 900|Supervising Sound...|Christopher Boyes|\n",
      "|   19995|Avatar|539c4a4cc3a36810c...|Production|     1|1262|             Casting|        Mali Finn|\n",
      "+--------+------+--------------------+----------+------+----+--------------------+-----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "crew_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "5fea0ca8-ea13-4d20-9313-c6f4dde05805",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Fill null values as mentioned in expected_output.ipynb file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "b5664934-e09b-4023-98aa-03487b3f2087",
   "metadata": {},
   "outputs": [],
   "source": [
    "cast_df = cast_df.na.fill(\"0000000000\",[\"credit_id\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "9f241521-63f1-43ca-8ead-4f77961ce819",
   "metadata": {},
   "outputs": [],
   "source": [
    "crew_df = crew_df.na.fill(\"0000000000\",[\"credit_id\"])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a4b5ab18-4056-422d-b0ed-c0a89d9981b3",
   "metadata": {},
   "source": [
    "#### MOVIES"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "6b8d2a93-3174-42b2-95ce-737243ac22d2",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Select necessary columns for expected output.\n",
    "df_movies_selected = df_movies.select(\"id\",\"title\", \"budget\", \"homepage\", \"original_language\", \"original_title\", \"overview\", \"popularity\", \"release_date\", \"revenue\", \"runtime\", \"status\", \\\n",
    "                 \"tagline\", \"vote_average\", \"vote_count\").withColumnRenamed(\"id\", \"movie_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "a5791811-137b-4e78-8948-302729c7cbc2",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movie_id: long (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- budget: long (nullable = true)\n",
      " |-- homepage: string (nullable = true)\n",
      " |-- original_language: string (nullable = true)\n",
      " |-- original_title: string (nullable = true)\n",
      " |-- overview: string (nullable = true)\n",
      " |-- popularity: double (nullable = true)\n",
      " |-- release_date: string (nullable = true)\n",
      " |-- revenue: long (nullable = true)\n",
      " |-- runtime: double (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- tagline: string (nullable = true)\n",
      " |-- vote_average: double (nullable = true)\n",
      " |-- vote_count: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_movies_selected.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "a5e05ccb-b1bb-4c58-9ffa-4d708444f7e0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Change the data types using cast() function.\n",
    "movies_df = df_movies_selected.withColumn(\"movie_id\", F.col(\"movie_id\").cast(StringType())) \\\n",
    ".withColumn(\"budget\", F.col(\"budget\").cast(DoubleType())) \\\n",
    ".withColumn(\"popularity\", F.col(\"budget\").cast(FloatType())) \\\n",
    ".withColumn(\"release_date\", F.to_date(\"release_date\", \"yyyy-MM-dd\")) \\\n",
    ".withColumn(\"revenue\", F.col(\"budget\").cast(DoubleType())) \\\n",
    ".withColumn(\"runtime\", F.col(\"budget\").cast(IntegerType())) \\\n",
    ".withColumn(\"vote_average\", F.col(\"vote_average\").cast(FloatType())) \\\n",
    ".withColumn(\"vote_count\", F.col(\"vote_count\").cast(IntegerType()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "fd75aff7-b445-4c78-9249-f384bb468857",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------------+------+--------------------+-----------------+--------------------+--------------------+----------+------------+-------+---------+--------+--------------------+------------+----------+\n",
      "|movie_id|               title|budget|            homepage|original_language|      original_title|            overview|popularity|release_date|revenue|  runtime|  status|             tagline|vote_average|vote_count|\n",
      "+--------+--------------------+------+--------------------+-----------------+--------------------+--------------------+----------+------------+-------+---------+--------+--------------------+------------+----------+\n",
      "|   19995|              Avatar|2.37E8|http://www.avatar...|               en|              Avatar|In the 22nd centu...|    2.37E8|  2009-12-10| 2.37E8|237000000|Released|Enter the World o...|         7.2|     11800|\n",
      "|     285|Pirates of the Ca...| 3.0E8|http://disney.go....|               en|Pirates of the Ca...|Captain Barbossa,...|     3.0E8|  2007-05-19|  3.0E8|300000000|Released|At the end of the...|         6.9|      4500|\n",
      "|  206647|             Spectre|2.45E8|http://www.sonypi...|               en|             Spectre|A cryptic message...|    2.45E8|  2015-10-26| 2.45E8|245000000|Released|A Plan No One Esc...|         6.3|      4466|\n",
      "|   49026|The Dark Knight R...| 2.5E8|http://www.thedar...|               en|The Dark Knight R...|Following the dea...|     2.5E8|  2012-07-16|  2.5E8|250000000|Released|     The Legend Ends|         7.6|      9106|\n",
      "|   49529|         John Carter| 2.6E8|http://movies.dis...|               en|         John Carter|John Carter is a ...|     2.6E8|  2012-03-07|  2.6E8|260000000|Released|Lost in our world...|         6.1|      2124|\n",
      "+--------+--------------------+------+--------------------+-----------------+--------------------+--------------------+----------+------------+-------+---------+--------+--------------------+------------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movies_df.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "8fbe1888-681f-47a1-9630-da21c2ad36f5",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movie_id: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- budget: double (nullable = true)\n",
      " |-- homepage: string (nullable = true)\n",
      " |-- original_language: string (nullable = true)\n",
      " |-- original_title: string (nullable = true)\n",
      " |-- overview: string (nullable = true)\n",
      " |-- popularity: float (nullable = true)\n",
      " |-- release_date: date (nullable = true)\n",
      " |-- revenue: double (nullable = true)\n",
      " |-- runtime: integer (nullable = true)\n",
      " |-- status: string (nullable = true)\n",
      " |-- tagline: string (nullable = true)\n",
      " |-- vote_average: float (nullable = true)\n",
      " |-- vote_count: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movies_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "75dd435d-dd99-4391-9e45-33e0e02fe615",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### GENRES, KEYWORDS, PRODUCTION COMPANY"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "50d690d7-df1c-456c-80c0-d7ee78c153e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# This part is similar to previous steps.\n",
    "# Before starting you can check the \"DATATYPE CONTROL\" section."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "a3ee3d2a-a514-4072-87d0-b8adcb7285a4",
   "metadata": {},
   "outputs": [],
   "source": [
    "schema = (\n",
    "    ArrayType(\n",
    "        StructType([\n",
    "            StructField(\"id\", IntegerType()),\n",
    "            StructField(\"name\", StringType()),\n",
    "                  ])\n",
    "            )\n",
    "         )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "21fda0bc-6575-4977-bc82-85523aea8d23",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "genres\n",
      "keywords\n",
      "production_companies\n"
     ]
    }
   ],
   "source": [
    "for col in [\"genres\", \"keywords\", \"production_companies\"]:\n",
    "    print(col)\n",
    "    gkp_df_raw =df_movies.select(\"id\",col).withColumnRenamed(\"id\", \"movie_id\")\n",
    "    gkp_df_raw = gkp_df_raw.withColumn(col, F.from_json(col, schema))\n",
    "    gkp_df = gkp_df_raw.select(\"movie_id\", F.explode_outer(col)).select(\"movie_id\",\"col.id\",\"col.name\").withColumn(\"movie_id\", F.col(\"movie_id\").cast(StringType()))\n",
    "    gkp_df = gkp_df.na.fill(value=-9999,subset=[\"id\"])\n",
    "    if col == \"genres\":\n",
    "        genres = gkp_df\n",
    "    elif col == \"keywords\":\n",
    "        keywords = gkp_df\n",
    "    else :\n",
    "        prod_comp = gkp_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "id": "87cc632d-a823-41a0-9d40-8888ef5a0807",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---+---------------+\n",
      "|movie_id| id|           name|\n",
      "+--------+---+---------------+\n",
      "|   19995| 28|         Action|\n",
      "|   19995| 12|      Adventure|\n",
      "|   19995| 14|        Fantasy|\n",
      "|   19995|878|Science Fiction|\n",
      "|     285| 12|      Adventure|\n",
      "+--------+---+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "genres.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "22a62e5a-3cb8-4681-892f-e263a93213d7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----+-------------+\n",
      "|movie_id|  id|         name|\n",
      "+--------+----+-------------+\n",
      "|   19995|1463|culture clash|\n",
      "|   19995|2964|       future|\n",
      "|   19995|3386|    space war|\n",
      "|   19995|3388| space colony|\n",
      "|   19995|3679|      society|\n",
      "+--------+----+-------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "keywords.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "c156a37f-e77b-4513-ba0f-809f6442c5a9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---+--------------------+\n",
      "|movie_id| id|                name|\n",
      "+--------+---+--------------------+\n",
      "|   19995|289|Ingenious Film Pa...|\n",
      "|   19995|306|Twentieth Century...|\n",
      "|   19995|444|  Dune Entertainment|\n",
      "|   19995|574|Lightstorm Entert...|\n",
      "|     285|  2|Walt Disney Pictures|\n",
      "+--------+---+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "prod_comp.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "91ec346f-b21d-477f-bd35-13c1a6ba13c9",
   "metadata": {},
   "source": [
    "#### PRODUCTION COUNTRIES, SPOKEN LANGUAGES"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "a5b1f6bf-cd3e-4b8f-940d-3e5abdd14ccd",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "for col in [\"production_countries\", \"spoken_languages\"]:\n",
    "    ps_df_raw = df_movies.select(\"id\",col).withColumnRenamed(\"id\", \"movie_id\") \n",
    "    if col == \"production_countries\" :\n",
    "        sub_col = \"iso_3166_1\"\n",
    "    else :\n",
    "        sub_col = \"iso_639_1\"\n",
    "    schema = (\n",
    "    ArrayType(\n",
    "        StructType([\n",
    "            StructField(sub_col, StringType()),\n",
    "            StructField(\"name\", StringType()),\n",
    "                  ])\n",
    "            )\n",
    "         )\n",
    "    ps_df_raw = ps_df_raw.withColumn(col, F.from_json(col, schema))\n",
    "    ps_df = ps_df_raw.select(\"movie_id\", F.explode_outer(col)).select(\"movie_id\",f\"col.{sub_col}\",\"col.name\").withColumn(\"movie_id\", F.col(\"movie_id\").cast(StringType()))\n",
    "    ps_df = ps_df.na.fill(\"XX\",[sub_col])\n",
    "    if col == \"production_countries\":\n",
    "        prod_coun = ps_df\n",
    "    else :\n",
    "        spo_lang = ps_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "id": "985bd88c-774d-4213-b313-f222cce19257",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----------+--------------------+\n",
      "|movie_id|iso_3166_1|                name|\n",
      "+--------+----------+--------------------+\n",
      "|   19995|        US|United States of ...|\n",
      "|   19995|        GB|      United Kingdom|\n",
      "|     285|        US|United States of ...|\n",
      "|  206647|        GB|      United Kingdom|\n",
      "|  206647|        US|United States of ...|\n",
      "+--------+----------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "prod_coun.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "id": "cb69d7b6-4dd7-4403-b1b2-e504f94c0359",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+---------+--------+\n",
      "|movie_id|iso_639_1|    name|\n",
      "+--------+---------+--------+\n",
      "|   19995|       en| English|\n",
      "|   19995|       es| Español|\n",
      "|     285|       en| English|\n",
      "|  206647|       fr|Français|\n",
      "|  206647|       en| English|\n",
      "+--------+---------+--------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spo_lang.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4ef766ef-96b7-4db0-b1ae-790cfc9a36d8",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true,
    "tags": []
   },
   "source": [
    "#### DATATYPE CONTROL"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "c6536e98-91fb-4626-862c-3b254aed897e",
   "metadata": {},
   "outputs": [],
   "source": [
    "genres_df_raw = df_movies.select(\"id\",\"genres\").withColumnRenamed(\"id\", \"movie_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "id": "14042300-283a-4359-a9bf-65cee1cfbee4",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(movie_id=19995, genres='[{\"id\": 28, \"name\": \"Action\"}, {\"id\": 12, \"name\": \"Adventure\"}, {\"id\": 14, \"name\": \"Fantasy\"}, {\"id\": 878, \"name\": \"Science Fiction\"}]')"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "genres_df_raw.collect()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "7f680640-48de-4377-bc88-598fff788b1c",
   "metadata": {},
   "outputs": [],
   "source": [
    "keywords_df_raw = df_movies.select(\"id\",\"keywords\").withColumnRenamed(\"id\", \"movie_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "id": "7c36e1c1-7955-4442-9126-fb71b43fbe96",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(movie_id=19995, keywords='[{\"id\": 1463, \"name\": \"culture clash\"}, {\"id\": 2964, \"name\": \"future\"}, {\"id\": 3386, \"name\": \"space war\"}, {\"id\": 3388, \"name\": \"space colony\"}, {\"id\": 3679, \"name\": \"society\"}, {\"id\": 3801, \"name\": \"space travel\"}, {\"id\": 9685, \"name\": \"futuristic\"}, {\"id\": 9840, \"name\": \"romance\"}, {\"id\": 9882, \"name\": \"space\"}, {\"id\": 9951, \"name\": \"alien\"}, {\"id\": 10148, \"name\": \"tribe\"}, {\"id\": 10158, \"name\": \"alien planet\"}, {\"id\": 10987, \"name\": \"cgi\"}, {\"id\": 11399, \"name\": \"marine\"}, {\"id\": 13065, \"name\": \"soldier\"}, {\"id\": 14643, \"name\": \"battle\"}, {\"id\": 14720, \"name\": \"love affair\"}, {\"id\": 165431, \"name\": \"anti war\"}, {\"id\": 193554, \"name\": \"power relations\"}, {\"id\": 206690, \"name\": \"mind and soul\"}, {\"id\": 209714, \"name\": \"3d\"}]')"
      ]
     },
     "execution_count": 40,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "keywords_df_raw.collect()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "b744f066-f9be-4a95-a72f-dca533516793",
   "metadata": {},
   "outputs": [],
   "source": [
    "pro_df_raw = df_movies.select(\"id\",\"production_companies\").withColumnRenamed(\"id\", \"movie_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "9035fdbc-ea86-497a-9c9e-be973c075359",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(movie_id=19995, production_companies='[{\"name\": \"Ingenious Film Partners\", \"id\": 289}, {\"name\": \"Twentieth Century Fox Film Corporation\", \"id\": 306}, {\"name\": \"Dune Entertainment\", \"id\": 444}, {\"name\": \"Lightstorm Entertainment\", \"id\": 574}]')"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pro_df_raw.collect()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "df82bf39-f48f-4518-a82a-f0d420534ae4",
   "metadata": {},
   "outputs": [],
   "source": [
    "pro_count_df_raw = df_movies.select(\"id\",\"production_countries\").withColumnRenamed(\"id\", \"movie_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "8d9e8519-0efd-4c50-a120-6e13e3282143",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(movie_id=19995, production_countries='[{\"iso_3166_1\": \"US\", \"name\": \"United States of America\"}, {\"iso_3166_1\": \"GB\", \"name\": \"United Kingdom\"}]')"
      ]
     },
     "execution_count": 44,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pro_count_df_raw .collect()[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "id": "66c7382f-6fae-4143-8d0d-a1015503a134",
   "metadata": {},
   "outputs": [],
   "source": [
    "spo_lan_df_raw = df_movies.select(\"id\",\"spoken_languages\").withColumnRenamed(\"id\", \"movie_id\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "id": "f458807a-10b5-4b42-bdac-7d7f10674691",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Row(movie_id=19995, spoken_languages='[{\"iso_639_1\": \"en\", \"name\": \"English\"}, {\"iso_639_1\": \"es\", \"name\": \"Espa\\\\u00f1ol\"}]')"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spo_lan_df_raw .collect()[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bac7817f-2431-4996-b904-09c97101116f",
   "metadata": {},
   "source": [
    "#### STOP"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "80c2ffb2-b2e4-4212-bf0e-290a8d57b842",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9047244a-8905-491d-b211-94de14224a6e",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "venvspark",
   "language": "python",
   "name": "venvspark"
  },
  "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}