Pyspark_AirBnB_End2End.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "e0b16b58",
   "metadata": {},
   "source": [
    "### Motivation\n",
    "\n",
    "- Creating a pipeline to move the data from simple files to database using Pyspark. \n",
    "\n",
    "- Create the tables in Database \n",
    "\n",
    "- Connect to the database pull the data for analysis using Pyspark\n",
    "\n",
    "- After analysis write the updated table / new table to database\n",
    "\n",
    "- Write the final analysis report to csv file and send to local file system"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e6ab6b3c",
   "metadata": {},
   "source": [
    "In order to analysis the data, it has to be in the memory. Spark creates JVM in which the data can be loaded and processed. Pyspark provides multiple routes for loading the data into the Context. From the files inside the folder. Connecting with database server like postgres, and loading the tables from there. It also allows to read the file directly using sparkContext \n",
    "\n",
    "Pyspark allows to move the analysis to SQL engine / Hive engine lying below spark context, using the CreateTempView. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "a5fc9f12",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark import SparkContext,SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *\n",
    "from pyspark.sql.window import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "427eb32b",
   "metadata": {},
   "outputs": [],
   "source": [
    "airbnb_root = \"/home/****/spark-warehouse/airbnb/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "1766a6b0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/11/23 11:14:07 WARN Utils: Your hostname, codeStation resolves to a loopback address: 127.0.1.1; using 192.168.126.83 instead (on interface wlo1)\n",
      "22/11/23 11:14:07 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "22/11/23 11:14:09 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession.builder \\\n",
    "        .appName('AirBnB Session') \\\n",
    "        .config('spark.jars','/usr/share/java/postgresql-42.2.26.jar') \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "867b9d2e",
   "metadata": {},
   "outputs": [],
   "source": [
    "sparkCon = spark.sparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "7abfe1c5",
   "metadata": {},
   "outputs": [],
   "source": [
    "#The below url references the sql server running in local machiew with Airbnb database\n",
    "url = \"jdbc:postgresql://localhost:5432/airbnb\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "6f9c4034",
   "metadata": {},
   "outputs": [],
   "source": [
    "calendar = sparkCon.textFile(airbnb_root + \"calendar.csv\") #This method seems redundant \n",
    "#The above method still needs to be reviewed with different kind of datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "ef57c2c0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#HDFS the databases, tables are just folders. In the local file-system NTFS/Fat32 also they\n",
    "#exists the same way\n",
    "spark.sql(\"CREATE DATABASE IF NOT EXISTS airbnb_pyspark\")\n",
    "spark.sql(\"use airbnb_pyspark\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "362cfb28",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----------+\n",
      "|namespace|tableName|isTemporary|\n",
      "+---------+---------+-----------+\n",
      "+---------+---------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#The tables shows below are inside the Pyspark environment. After I bring all the necessary\n",
    "#data inside pyspark environment\n",
    "spark.sql(\"SHOW TABLES\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "9f07cf71",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#Creating the spark RDD Dataframes to get the data into pyspark\n",
    "calendarTab = spark.read.csv(airbnb_root+\"calendar.csv\",inferSchema=True,header=True)\n",
    "#Initialising the similar table inside the Pyspark environment for SQL based analysis\n",
    "calendarTab.createOrReplaceTempView(\"airbnb_pyspark.calendar\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "3be76f2d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------------------+---------+-----+\n",
      "|listing_id|               date|available|price|\n",
      "+----------+-------------------+---------+-----+\n",
      "|      2818|2019-12-05 00:00:00|        f| null|\n",
      "|     73208|2019-08-30 00:00:00|        f| null|\n",
      "|     73208|2019-08-29 00:00:00|        f| null|\n",
      "|     73208|2019-08-28 00:00:00|        f| null|\n",
      "|     73208|2019-08-27 00:00:00|        f| null|\n",
      "+----------+-------------------+---------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "calendarTab.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1d1d2bf6",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Note that the below activity is required only once. With help of overwrite mode one can \n",
    "#keep re-doing it. It will involve calls to Database connectivity\n",
    "#Initialising the database connection and writing the table to it\n",
    "calendarTab.write.format(\"jdbc\") \\\n",
    "        .option(\"url\",\"jdbc:postgresql://localhost:5432/airbnb\") \\\n",
    "        .option(\"dbtable\",\"calendar\") \\\n",
    "        .option(\"user\",\"postgres\") \\\n",
    "        .option(\"password\",1234) \\\n",
    "        .option(\"driver\",\"org.postgresql.Driver\") \\\n",
    "        .save(mode='overwrite')\n",
    "#Writing the file to the database and making it permanent\n",
    "calendarTab.write.saveAsTable(\"calendar\",mode='overwrite')\n",
    "#Checkout the files, they are super compressed into Parquets using Snappy Algorithms"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "20cc667a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/11/22 20:58:03 WARN HadoopFSUtils: The directory file:/run/media/solverbot/repoA/gitFolders/moreDE/spark-warehouse/airbnb_pyspark.db/calendartab was not found. Was it deleted very recently?\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "DataFrame[]"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#The below command deletes the file in the underlying file system\n",
    "spark.sql(\"DROP TABLE calendartab\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "77a1e9e5",
   "metadata": {},
   "outputs": [],
   "source": [
    "listingTab = spark.read.csv(airbnb_root+\"listings.csv\",inferSchema=True,header=True)\n",
    "#Initialising the database loading \n",
    "listingTab.createOrReplaceTempView(\"listings\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "1edb09fe",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: string (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- host_id: string (nullable = true)\n",
      " |-- host_name: string (nullable = true)\n",
      " |-- neighbourhood_group: string (nullable = true)\n",
      " |-- neighbourhood: string (nullable = true)\n",
      " |-- latitude: string (nullable = true)\n",
      " |-- longitude: string (nullable = true)\n",
      " |-- room_type: string (nullable = true)\n",
      " |-- price: string (nullable = true)\n",
      " |-- minimum_nights: integer (nullable = true)\n",
      " |-- number_of_reviews: string (nullable = true)\n",
      " |-- last_review: string (nullable = true)\n",
      " |-- reviews_per_month: string (nullable = true)\n",
      " |-- calculated_host_listings_count: double (nullable = true)\n",
      " |-- availability_365: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "listingTab.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "6d65d452",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------------------------+-----+----------------+\n",
      "|name                                             |price|availability_365|\n",
      "+-------------------------------------------------+-----+----------------+\n",
      "|Quiet Garden View Room & Super Fast WiFi         |59   |44              |\n",
      "|Quiet apt near center, great view                |160  |47              |\n",
      "|100%Centre-Studio 1 Private Floor/Bathroom       |80   |198             |\n",
      "|Lovely apt in City Centre (Jordaan)              |125  |141             |\n",
      "|Romantic, stylish B&B houseboat in canal district|150  |199             |\n",
      "+-------------------------------------------------+-----+----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Taking only part of the columns that is interesting using Spark Dataframe method\n",
    "listingTab.select(\"name\",\"price\",\"availability_365\").show(5,truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "7f007078",
   "metadata": {},
   "outputs": [],
   "source": [
    "listings_truncated = spark.sql(\"SELECT name, price, availability_365 FROM listings\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "33001b51",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------------------------+-----+----------------+\n",
      "|name                                    |price|availability_365|\n",
      "+----------------------------------------+-----+----------------+\n",
      "|Quiet Garden View Room & Super Fast WiFi|59   |44              |\n",
      "|Quiet apt near center, great view       |160  |47              |\n",
      "+----------------------------------------+-----+----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "listings_truncated.show(2,truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "ba897aea",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- name: string (nullable = true)\n",
      " |-- price: string (nullable = true)\n",
      " |-- availability_365: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "listings_truncated.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "35641d32",
   "metadata": {},
   "outputs": [],
   "source": [
    "#This will create a folder and write the CSV files inside the listings_tab folder, \n",
    "listings_truncated.write.csv(\"listings_tab\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5d2a47ef",
   "metadata": {},
   "outputs": [],
   "source": [
    "#This will connect to the external Postgres Data base and create the table \n",
    "#Listings_truncated and write the data, \n",
    "listing_truncated.write.format(\"jdbc\") \\\n",
    "        .option(\"url\",\"jdbc:postgresql://localhost:5432/airbnb\") \\\n",
    "        .option(\"dbtable\",\"listings_truncated\") \\\n",
    "        .option(\"user\",\"postgres\") \\\n",
    "        .option(\"password\",1234) \\\n",
    "        .option(\"driver\",\"org.postgresql.Driver\") \\\n",
    "        .save(mode='overwrite')\n",
    "listingTab.write.saveAsTable(\"listings\",mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ae1a980",
   "metadata": {},
   "outputs": [],
   "source": [
    "#This command will store the files in local environment as database\n",
    "listings_truncated.write.saveAsTable(\"listings_truncated\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "44656476",
   "metadata": {},
   "outputs": [],
   "source": [
    "reviewsTab.write.format('jdbc') \\\n",
    "    .option(\"url\",url) \\\n",
    "    .option('dbtable','reviews') \\\n",
    "    .option('user','postgres') \\\n",
    "    .option('password',1234) \\\n",
    "    .option('driver','org.postgresql.Driver') \\\n",
    "    .save(mode='ignore')\n",
    "reviewsTab.write.saveAsTable('reviews')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "e52969e9",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 7:>                                                          (0 + 2) / 2]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#Creating the reviews tables in postgres database, local folder system and in pyspark env\n",
    "reviewsTab = spark.read.csv(airbnb_root+\"reviews.csv\",inferSchema=True,header=True)\n",
    "reviewsTab.createOrReplaceTempView(\"reviews\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "333b90e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "reviewsDetails.write.format('jdbc').option('url',url).option(\"dbtable\",'reviews_details') \\\n",
    "    .option(\"user\",\"postgres\").option(\"password\",1234).option(\"driver\",\"org.postgresql.Driver\") \\\n",
    "    .save(mode='ignore')\n",
    "reviewsDetails.write.saveAsTable(\"review_tables\") #permanent to folder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "f18cdee9",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#Creating the review details tables in postgres database, local folder system and in pyspark env\n",
    "reviewsDetails = spark.read.csv(airbnb_root+\"reviews_details.csv\",inferSchema=True,header=True)\n",
    "reviewsDetails.createOrReplaceTempView(\"review_details\") #temporary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "76267de6",
   "metadata": {},
   "outputs": [],
   "source": [
    "neighbourhood.write.format('jdbc').option('url',url).option(\"dbtable\",'neighbourhoods') \\\n",
    "    .option('user','postgres').option('password',1234).option('driver','org.postgresql.Driver') \\\n",
    "    .save(mode='ignore')\n",
    "neighbourhood.write.saveAsTable(\"neighbourhoods\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "2cd8cba1",
   "metadata": {},
   "outputs": [],
   "source": [
    "neighbourhood = spark.read.csv(airbnb_root+\"neighbourhoods.csv\",inferSchema=True, header=True)\n",
    "neighbourhood.createOrReplaceTempView(\"neighbourhoods\") #used for the spark sql activity\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "id": "c4cd378b",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on method json in module pyspark.sql.readwriter:\n",
      "\n",
      "json(path: Union[str, List[str], pyspark.rdd.RDD[str]], schema: Union[pyspark.sql.types.StructType, str, NoneType] = None, primitivesAsString: Union[bool, str, NoneType] = None, prefersDecimal: Union[bool, str, NoneType] = None, allowComments: Union[bool, str, NoneType] = None, allowUnquotedFieldNames: Union[bool, str, NoneType] = None, allowSingleQuotes: Union[bool, str, NoneType] = None, allowNumericLeadingZero: Union[bool, str, NoneType] = None, allowBackslashEscapingAnyCharacter: Union[bool, str, NoneType] = None, mode: Optional[str] = None, columnNameOfCorruptRecord: Optional[str] = None, dateFormat: Optional[str] = None, timestampFormat: Optional[str] = None, multiLine: Union[bool, str, NoneType] = None, allowUnquotedControlChars: Union[bool, str, NoneType] = None, lineSep: Optional[str] = None, samplingRatio: Union[str, float, NoneType] = None, dropFieldIfAllNull: Union[bool, str, NoneType] = None, encoding: Optional[str] = None, locale: Optional[str] = None, pathGlobFilter: Union[bool, str, NoneType] = None, recursiveFileLookup: Union[bool, str, NoneType] = None, modifiedBefore: Union[bool, str, NoneType] = None, modifiedAfter: Union[bool, str, NoneType] = None, allowNonNumericNumbers: Union[bool, str, NoneType] = None) -> 'DataFrame' method of pyspark.sql.readwriter.DataFrameReader instance\n",
      "    Loads JSON files and returns the results as a :class:`DataFrame`.\n",
      "    \n",
      "    `JSON Lines <http://jsonlines.org/>`_ (newline-delimited JSON) is supported by default.\n",
      "    For JSON (one record per file), set the ``multiLine`` parameter to ``true``.\n",
      "    \n",
      "    If the ``schema`` parameter is not specified, this function goes\n",
      "    through the input once to determine the input schema.\n",
      "    \n",
      "    .. versionadded:: 1.4.0\n",
      "    \n",
      "    Parameters\n",
      "    ----------\n",
      "    path : str, list or :class:`RDD`\n",
      "        string represents path to the JSON dataset, or a list of paths,\n",
      "        or RDD of Strings storing JSON objects.\n",
      "    schema : :class:`pyspark.sql.types.StructType` or str, optional\n",
      "        an optional :class:`pyspark.sql.types.StructType` for the input schema or\n",
      "        a DDL-formatted string (For example ``col0 INT, col1 DOUBLE``).\n",
      "    \n",
      "    Other Parameters\n",
      "    ----------------\n",
      "    Extra options\n",
      "        For the extra options, refer to\n",
      "        `Data Source Option <https://spark.apache.org/docs/latest/sql-data-sources-json.html#data-source-option>`_\n",
      "        in the version you use.\n",
      "    \n",
      "        .. # noqa\n",
      "    \n",
      "    Examples\n",
      "    --------\n",
      "    >>> df1 = spark.read.json('python/test_support/sql/people.json')\n",
      "    >>> df1.dtypes\n",
      "    [('age', 'bigint'), ('name', 'string')]\n",
      "    >>> rdd = sc.textFile('python/test_support/sql/people.json')\n",
      "    >>> df2 = spark.read.json(rdd)\n",
      "    >>> df2.dtypes\n",
      "    [('age', 'bigint'), ('name', 'string')]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(spark.read.json)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "id": "337060c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Reading the geojson file requires additional libraries from Apache, called Apache Sedona\n",
    "#Will explore and update it soon.\n",
    "neighbourhoodjson = spark.read.json(airbnb_root+\"neighbourhoods.geojson\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "id": "78dc4d14",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- features: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- geometry: struct (nullable = true)\n",
      " |    |    |    |-- coordinates: array (nullable = true)\n",
      " |    |    |    |    |-- element: array (containsNull = true)\n",
      " |    |    |    |    |    |-- element: array (containsNull = true)\n",
      " |    |    |    |    |    |    |-- element: array (containsNull = true)\n",
      " |    |    |    |    |    |    |    |-- element: double (containsNull = true)\n",
      " |    |    |    |-- type: string (nullable = true)\n",
      " |    |    |-- properties: struct (nullable = true)\n",
      " |    |    |    |-- neighbourhood: string (nullable = true)\n",
      " |    |    |    |-- neighbourhood_group: string (nullable = true)\n",
      " |    |    |-- type: string (nullable = true)\n",
      " |-- type: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "neighbourhoodjson.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "f0d045ac",
   "metadata": {},
   "outputs": [],
   "source": [
    "#When we have bulk data, that is spanning many files. Pyspark makes it amazingly easy to \n",
    "#pull the data into the pyspark environment, process it, move it to multiple output points.\n",
    "ytCSV = \"/home/solverbot/Desktop/ytDE/csvfiles\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "a7fc558d",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#There are bunch of files inside the above path. Going to check what happens\n",
    "#When I pass it through spark.read.csv\n",
    "ytvidPy = spark.read.csv(ytCSV,inferSchema=True, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "id": "92651e88",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------------+--------------------+-------------+-----------+--------------------+--------------------+------+-----+--------+-------------+--------------------+-----------------+----------------+----------------------+--------------------+\n",
      "|   video_id|trending_date|               title|channel_title|category_id|        publish_time|                tags| views|likes|dislikes|comment_count|      thumbnail_link|comments_disabled|ratings_disabled|video_error_or_removed|         description|\n",
      "+-----------+-------------+--------------------+-------------+-----------+--------------------+--------------------+------+-----+--------+-------------+--------------------+-----------------+----------------+----------------------+--------------------+\n",
      "|gDuslQ9avLc|     17.14.11|Захар и Полина уч...|    Т—Ж БОГАЧ|         22|2017-11-13T09:09:...|\"захар и полина|\"...| 62408|  334|     190|           50|https://i.ytimg.c...|            FALSE|           FALSE|                 FALSE|Знакомьтесь, это ...|\n",
      "|AOCJIFEA_jE|     17.14.11|Биржа Мемов #29. ...| Druzhko Show|         22|2017-11-13T17:32:...|\"биржа мемов|\"\"ле...|330043|43841|    2244|         2977|https://i.ytimg.c...|            FALSE|           FALSE|                 FALSE|В 29 выпуске Друж...|\n",
      "+-----------+-------------+--------------------+-------------+-----------+--------------------+--------------------+------+-----+--------+-------------+--------------------+-----------------+----------------+----------------------+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ytvidPy.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "id": "6f36d13b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- video_id: string (nullable = true)\n",
      " |-- trending_date: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- channel_title: string (nullable = true)\n",
      " |-- category_id: string (nullable = true)\n",
      " |-- publish_time: string (nullable = true)\n",
      " |-- tags: string (nullable = true)\n",
      " |-- views: string (nullable = true)\n",
      " |-- likes: string (nullable = true)\n",
      " |-- dislikes: string (nullable = true)\n",
      " |-- comment_count: string (nullable = true)\n",
      " |-- thumbnail_link: string (nullable = true)\n",
      " |-- comments_disabled: string (nullable = true)\n",
      " |-- ratings_disabled: string (nullable = true)\n",
      " |-- video_error_or_removed: string (nullable = true)\n",
      " |-- description: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ytvidPy.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "id": "afc6ac8a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "416869"
      ]
     },
     "execution_count": 72,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#All the 10 files each averaging 40M is loaded into single dataframe\n",
    "ytvidPy.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "19a6236c",
   "metadata": {},
   "outputs": [],
   "source": [
    "ytvidPy.write.format('jdbc').option('')\n",
    "#The below save to parquet compressed the files to 270 MB, almost 50% compression \n",
    "ytvidPy.write.saveAsTable(\"youtubeVideos\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "eb386b40",
   "metadata": {},
   "outputs": [],
   "source": [
    "ytvidPy.createOrReplaceTempView(\"youtubeVideos\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "26d3d58e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#The same is not true for the json files though. They need additional processing\n",
    "ytJson = \"/home/solverbot/Desktop/ytDE/jsonfiles/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "id": "9dbcca74",
   "metadata": {},
   "outputs": [],
   "source": [
    "readingYtJson = spark.read.json(ytJson)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "id": "2d420eb2",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "ename": "AnalysisException",
     "evalue": "\nSince Spark 2.3, the queries from raw JSON/CSV files are disallowed when the\nreferenced columns only include the internal corrupt record column\n(named _corrupt_record by default). For example:\nspark.read.schema(schema).csv(file).filter($\"_corrupt_record\".isNotNull).count()\nand spark.read.schema(schema).csv(file).select(\"_corrupt_record\").show().\nInstead, you can cache or save the parsed results and then send the same query.\nFor example, val df = spark.read.schema(schema).csv(file).cache() and then\ndf.filter($\"_corrupt_record\".isNotNull).count().\n      ",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mAnalysisException\u001b[0m                         Traceback (most recent call last)",
      "Cell \u001b[0;32mIn [79], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[43mreadingYtJson\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mtake\u001b[49m\u001b[43m(\u001b[49m\u001b[38;5;241;43m2\u001b[39;49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/.local/lib/python3.10/site-packages/pyspark/sql/dataframe.py:868\u001b[0m, in \u001b[0;36mDataFrame.take\u001b[0;34m(self, num)\u001b[0m\n\u001b[1;32m    858\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mtake\u001b[39m(\u001b[38;5;28mself\u001b[39m, num: \u001b[38;5;28mint\u001b[39m) \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m>\u001b[39m List[Row]:\n\u001b[1;32m    859\u001b[0m     \u001b[38;5;124;03m\"\"\"Returns the first ``num`` rows as a :class:`list` of :class:`Row`.\u001b[39;00m\n\u001b[1;32m    860\u001b[0m \n\u001b[1;32m    861\u001b[0m \u001b[38;5;124;03m    .. versionadded:: 1.3.0\u001b[39;00m\n\u001b[0;32m   (...)\u001b[0m\n\u001b[1;32m    866\u001b[0m \u001b[38;5;124;03m    [Row(age=2, name='Alice'), Row(age=5, name='Bob')]\u001b[39;00m\n\u001b[1;32m    867\u001b[0m \u001b[38;5;124;03m    \"\"\"\u001b[39;00m\n\u001b[0;32m--> 868\u001b[0m     \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mlimit\u001b[49m\u001b[43m(\u001b[49m\u001b[43mnum\u001b[49m\u001b[43m)\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcollect\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[0;32m~/.local/lib/python3.10/site-packages/pyspark/sql/dataframe.py:817\u001b[0m, in \u001b[0;36mDataFrame.collect\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m    807\u001b[0m \u001b[38;5;124;03m\"\"\"Returns all the records as a list of :class:`Row`.\u001b[39;00m\n\u001b[1;32m    808\u001b[0m \n\u001b[1;32m    809\u001b[0m \u001b[38;5;124;03m.. versionadded:: 1.3.0\u001b[39;00m\n\u001b[0;32m   (...)\u001b[0m\n\u001b[1;32m    814\u001b[0m \u001b[38;5;124;03m[Row(age=2, name='Alice'), Row(age=5, name='Bob')]\u001b[39;00m\n\u001b[1;32m    815\u001b[0m \u001b[38;5;124;03m\"\"\"\u001b[39;00m\n\u001b[1;32m    816\u001b[0m \u001b[38;5;28;01mwith\u001b[39;00m SCCallSiteSync(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_sc):\n\u001b[0;32m--> 817\u001b[0m     sock_info \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_jdf\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcollectToPython\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m    818\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mlist\u001b[39m(_load_from_socket(sock_info, BatchedSerializer(CPickleSerializer())))\n",
      "File \u001b[0;32m~/.local/lib/python3.10/site-packages/py4j/java_gateway.py:1321\u001b[0m, in \u001b[0;36mJavaMember.__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m   1315\u001b[0m command \u001b[38;5;241m=\u001b[39m proto\u001b[38;5;241m.\u001b[39mCALL_COMMAND_NAME \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m   1316\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mcommand_header \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m   1317\u001b[0m     args_command \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m   1318\u001b[0m     proto\u001b[38;5;241m.\u001b[39mEND_COMMAND_PART\n\u001b[1;32m   1320\u001b[0m answer \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mgateway_client\u001b[38;5;241m.\u001b[39msend_command(command)\n\u001b[0;32m-> 1321\u001b[0m return_value \u001b[38;5;241m=\u001b[39m \u001b[43mget_return_value\u001b[49m\u001b[43m(\u001b[49m\n\u001b[1;32m   1322\u001b[0m \u001b[43m    \u001b[49m\u001b[43manswer\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mgateway_client\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mtarget_id\u001b[49m\u001b[43m,\u001b[49m\u001b[43m \u001b[49m\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mname\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m   1324\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m temp_arg \u001b[38;5;129;01min\u001b[39;00m temp_args:\n\u001b[1;32m   1325\u001b[0m     temp_arg\u001b[38;5;241m.\u001b[39m_detach()\n",
      "File \u001b[0;32m~/.local/lib/python3.10/site-packages/pyspark/sql/utils.py:196\u001b[0m, in \u001b[0;36mcapture_sql_exception.<locals>.deco\u001b[0;34m(*a, **kw)\u001b[0m\n\u001b[1;32m    192\u001b[0m converted \u001b[38;5;241m=\u001b[39m convert_exception(e\u001b[38;5;241m.\u001b[39mjava_exception)\n\u001b[1;32m    193\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(converted, UnknownException):\n\u001b[1;32m    194\u001b[0m     \u001b[38;5;66;03m# Hide where the exception came from that shows a non-Pythonic\u001b[39;00m\n\u001b[1;32m    195\u001b[0m     \u001b[38;5;66;03m# JVM exception message.\u001b[39;00m\n\u001b[0;32m--> 196\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m converted \u001b[38;5;28;01mfrom\u001b[39;00m \u001b[38;5;28mNone\u001b[39m\n\u001b[1;32m    197\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[1;32m    198\u001b[0m     \u001b[38;5;28;01mraise\u001b[39;00m\n",
      "\u001b[0;31mAnalysisException\u001b[0m: \nSince Spark 2.3, the queries from raw JSON/CSV files are disallowed when the\nreferenced columns only include the internal corrupt record column\n(named _corrupt_record by default). For example:\nspark.read.schema(schema).csv(file).filter($\"_corrupt_record\".isNotNull).count()\nand spark.read.schema(schema).csv(file).select(\"_corrupt_record\").show().\nInstead, you can cache or save the parsed results and then send the same query.\nFor example, val df = spark.read.schema(schema).csv(file).cache() and then\ndf.filter($\"_corrupt_record\".isNotNull).count().\n      "
     ]
    }
   ],
   "source": [
    "readingYtJson.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "id": "07760ec5",
   "metadata": {},
   "outputs": [],
   "source": [
    "caJson = spark.read.json(ytJson+\"CA_category_id.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "836f0bdc",
   "metadata": {},
   "outputs": [],
   "source": [
    "jsonRawCA= sparkCon.textFile(ytJson+\"CA_category_id.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "6006ceba",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "38536370",
   "metadata": {},
   "outputs": [],
   "source": [
    "jsonFile = open(ytJson+\"CA_category_id.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "22760c67",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.10.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}