BD/Assignment_2_Johnny.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Assignment 2\n",
    "\n",
    "#### First Name: Johnny\n",
    "#### Last Name: Núñez\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Load Data from JSON"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 197,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "from pyspark.sql import SparkSession \n",
    "spark = SparkSession.builder.appName(\"Twitter Analysis\")\\\n",
    ".getOrCreate()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 198,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_twitter = spark.read.json(\"corona_tweet_new.json\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 199,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- created_at: string (nullable = true)\n",
      " |-- favorite_count: long (nullable = true)\n",
      " |-- hashtags: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- in_reply_to_status_id: string (nullable = true)\n",
      " |-- in_reply_to_user_id_str: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- reply_count: long (nullable = true)\n",
      " |-- retweet_count: long (nullable = true)\n",
      " |-- source: string (nullable = true)\n",
      " |-- user: struct (nullable = true)\n",
      " |    |-- contributors_enabled: boolean (nullable = true)\n",
      " |    |-- created_at: string (nullable = true)\n",
      " |    |-- default_profile: boolean (nullable = true)\n",
      " |    |-- default_profile_image: boolean (nullable = true)\n",
      " |    |-- description: string (nullable = true)\n",
      " |    |-- favourites_count: long (nullable = true)\n",
      " |    |-- follow_request_sent: string (nullable = true)\n",
      " |    |-- followers_count: long (nullable = true)\n",
      " |    |-- following: string (nullable = true)\n",
      " |    |-- friends_count: long (nullable = true)\n",
      " |    |-- geo_enabled: boolean (nullable = true)\n",
      " |    |-- id: long (nullable = true)\n",
      " |    |-- id_str: string (nullable = true)\n",
      " |    |-- is_translator: boolean (nullable = true)\n",
      " |    |-- lang: string (nullable = true)\n",
      " |    |-- listed_count: long (nullable = true)\n",
      " |    |-- location: string (nullable = true)\n",
      " |    |-- name: string (nullable = true)\n",
      " |    |-- notifications: string (nullable = true)\n",
      " |    |-- profile_background_color: string (nullable = true)\n",
      " |    |-- profile_background_image_url: string (nullable = true)\n",
      " |    |-- profile_background_image_url_https: string (nullable = true)\n",
      " |    |-- profile_background_tile: boolean (nullable = true)\n",
      " |    |-- profile_banner_url: string (nullable = true)\n",
      " |    |-- profile_image_url: string (nullable = true)\n",
      " |    |-- profile_image_url_https: string (nullable = true)\n",
      " |    |-- profile_link_color: string (nullable = true)\n",
      " |    |-- profile_sidebar_border_color: string (nullable = true)\n",
      " |    |-- profile_sidebar_fill_color: string (nullable = true)\n",
      " |    |-- profile_text_color: string (nullable = true)\n",
      " |    |-- profile_use_background_image: boolean (nullable = true)\n",
      " |    |-- protected: boolean (nullable = true)\n",
      " |    |-- screen_name: string (nullable = true)\n",
      " |    |-- statuses_count: long (nullable = true)\n",
      " |    |-- time_zone: string (nullable = true)\n",
      " |    |-- translator_type: string (nullable = true)\n",
      " |    |-- url: string (nullable = true)\n",
      " |    |-- utc_offset: string (nullable = true)\n",
      " |    |-- verified: boolean (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_twitter.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 200,
   "metadata": {},
   "outputs": [],
   "source": [
    "### From the user nestec col select the following cols only id_str,followers_count,friends_count and created at \n",
    "# (2 points)\n",
    "from pyspark.sql.functions import col \n",
    "df_twitter=df_twitter.select(df_twitter.created_at,\n",
    "                             df_twitter.favorite_count,\n",
    "                             df_twitter.hashtags,\n",
    "                             df_twitter.id,\n",
    "                             df_twitter.in_reply_to_status_id,\n",
    "                             df_twitter.in_reply_to_user_id_str,\n",
    "                             df_twitter.location,\n",
    "                             df_twitter.reply_count,\n",
    "                             df_twitter.retweet_count,\n",
    "                             df_twitter.source,\n",
    "                             df_twitter.user.id_str.alias('user_id_str'),\n",
    "                             df_twitter.user.followers_count.alias('user_followers_count'),\n",
    "                             df_twitter.user.friends_count.alias('user_friends_count'),\n",
    "                             df_twitter.user.created_at.alias('user_created_at'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 201,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- created_at: string (nullable = true)\n",
      " |-- favorite_count: long (nullable = true)\n",
      " |-- hashtags: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- id: string (nullable = true)\n",
      " |-- in_reply_to_status_id: string (nullable = true)\n",
      " |-- in_reply_to_user_id_str: string (nullable = true)\n",
      " |-- location: string (nullable = true)\n",
      " |-- reply_count: long (nullable = true)\n",
      " |-- retweet_count: long (nullable = true)\n",
      " |-- source: string (nullable = true)\n",
      " |-- user_id_str: string (nullable = true)\n",
      " |-- user_followers_count: long (nullable = true)\n",
      " |-- user_friends_count: long (nullable = true)\n",
      " |-- user_created_at: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_twitter.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 202,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "15894"
      ]
     },
     "execution_count": 202,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Print the total count of number of records in df_twitter(1 point)\n",
    "df_twitter.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 203,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------------------+\n",
      "|   extracted_source|              source|\n",
      "+-------------------+--------------------+\n",
      "|    Twitter Web App|<a href=\"https://...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "|    Twitter Web App|<a href=\"https://...|\n",
      "| Twitter Web Client|<a href=\"http://t...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "| Twitter for iPhone|<a href=\"http://t...|\n",
      "| Twitter for iPhone|<a href=\"http://t...|\n",
      "|    Twitter Web App|<a href=\"https://...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "|    Twitter Web App|<a href=\"https://...|\n",
      "| Twitter for iPhone|<a href=\"http://t...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "| Twitter for iPhone|<a href=\"http://t...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "| Twitter for iPhone|<a href=\"http://t...|\n",
      "|Twitter for Android|<a href=\"http://t...|\n",
      "+-------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Extract the source lable from source col by droping the anchor tab and save it as another col named extracted_source\n",
    "# for example <a href=\"https://mobile.twitter.com\" rel=\"nofollow\">Twitter Web App</a> => Twitter Web App\n",
    "# you can use \"<a [^>]+>([^<]+)\" as regualr expresion and the group would be 1 for this regular expression.\n",
    "#(4 points)\n",
    "from pyspark.sql.functions import regexp_extract, col\n",
    "\n",
    "df_twitter=df_twitter.withColumn(\"extracted_source\",regexp_extract('source',r'<a [^>]+>([^<]+)',1))\n",
    "df_twitter.select(col('extracted_source'),col('source')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 204,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convert the DataFrame into RDD\n",
    "rdd_twitter=df_twitter.rdd.map(tuple)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 205,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create a temporay table in memory with name as twitter (1 point)\n",
    "df_twitter.createOrReplaceTempView(\"twitter\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Analyze Data\n",
    "\n",
    "#### You will be writing code to find the answer to the questions listed below using Just RDD, Using spark SQL \n",
    "\n",
    "- Analyze using RDD \n",
    "- Analyze using Dataframe without temp table \n",
    "- Analyze using spark.sql with temp table\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.1 Get total number of unique users (1 point for each type)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 206,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[created_at: string, favorite_count: bigint, hashtags: array<string>, id: string, in_reply_to_status_id: string, in_reply_to_user_id_str: string, location: string, reply_count: bigint, retweet_count: bigint, source: string, user_id_str: string, user_followers_count: bigint, user_friends_count: bigint, user_created_at: string, extracted_source: string]"
      ]
     },
     "execution_count": 206,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_twitter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 207,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "14094"
      ]
     },
     "execution_count": 207,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD\n",
    "rdd_twitter.map(lambda x: x[10]).distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 208,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "14094"
      ]
     },
     "execution_count": 208,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using DataFrame\n",
    "df_user_count = df_twitter.select('user_id_str').distinct()\n",
    "df_user_count.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 209,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------------+\n",
      "|count(DISTINCT user_id_str)|\n",
      "+---------------------------+\n",
      "|                      14094|\n",
      "+---------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table.\n",
    "spark.sql('SELECT COUNT(DISTINCT(user_id_str)) FROM twitter').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.2 Get count of user who have more than 1 tweet in the data (2 points)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 210,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1016"
      ]
     },
     "execution_count": 210,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD\n",
    "rdd_twitter.map(lambda x: (x[10],1)).reduceByKey(lambda x,y: x+y).filter(lambda x: x[1]>1).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 211,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1016"
      ]
     },
     "execution_count": 211,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import pyspark.sql.functions as F\n",
    "# Using DataFrame\n",
    "# Get count of user who have more than 1 tweet in the data in df_twitter\n",
    "df_twitter.groupBy('user_id_str').count().filter(F.col('count')>1).count()\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 212,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+\n",
      "|count(num_tweets)|\n",
      "+-----------------+\n",
      "|             1016|\n",
      "+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table.\n",
    "spark.sql(\"SELECT count(num_tweets) \\\n",
    "          FROM (\\\n",
    "            SELECT count(id) as num_tweets \\\n",
    "            FROM twitter \\\n",
    "            group by user_id_str\\\n",
    "          ) where num_tweets>1 \").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.3 Get total number unique extracted_source (1 point each)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 213,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "133"
      ]
     },
     "execution_count": 213,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD\n",
    "# Get total number unique extracted_source output: 133\n",
    "rdd_twitter.map(lambda x: x[14]).distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 214,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "133"
      ]
     },
     "execution_count": 214,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using DataFrame\n",
    "df_source_distinct = df_twitter.select('extracted_source').distinct()\n",
    "df_source_distinct.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 215,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------+\n",
      "|count(DISTINCT extracted_source)|\n",
      "+--------------------------------+\n",
      "|                             133|\n",
      "+--------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table.\n",
    "spark.sql(\"SELECT \\\n",
    "           COUNT(DISTINCT(extracted_source)) \\\n",
    "           FROM twitter\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.4 Get top 5 most used extracted_source"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 216,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('Twitter for Android', 6262),\n",
       " ('Twitter for iPhone', 5698),\n",
       " ('Twitter Web App', 2878),\n",
       " ('Twitter for iPad', 428),\n",
       " ('Twitter Web Client', 136)]"
      ]
     },
     "execution_count": 216,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD (5 points)\n",
    "rdd_twitter.map(lambda x: (x[14],1)).reduceByKey(lambda x,y: x+y).sortBy(lambda x: x[1],ascending=False).take(5)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 217,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+-----+\n",
      "|   extracted_source|count|\n",
      "+-------------------+-----+\n",
      "|Twitter for Android| 6262|\n",
      "| Twitter for iPhone| 5698|\n",
      "|    Twitter Web App| 2878|\n",
      "|   Twitter for iPad|  428|\n",
      "| Twitter Web Client|  136|\n",
      "+-------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using DataFrame (2 points) \n",
    "df_twitter.groupBy('extracted_source').count().orderBy(F.desc('count')).limit(5).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 218,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+------------+\n",
      "|   extracted_source|count_source|\n",
      "+-------------------+------------+\n",
      "|Twitter for Android|        6262|\n",
      "| Twitter for iPhone|        5698|\n",
      "|    Twitter Web App|        2878|\n",
      "|   Twitter for iPad|         428|\n",
      "| Twitter Web Client|         136|\n",
      "+-------------------+------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table. (2 points)\n",
    "spark.sql(\"SELECT \\\n",
    "          extracted_source, \\\n",
    "          COUNT(extracted_source) as count_source \\\n",
    "          FROM twitter \\\n",
    "          GROUP BY extracted_source \\\n",
    "          ORDER BY count_source DESC\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.5 Get count of distinct hastags used ( 5 point each) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 219,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1215"
      ]
     },
     "execution_count": 219,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD \n",
    "rdd_twitter.flatMap(lambda x: x[2]).distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 220,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1215"
      ]
     },
     "execution_count": 220,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using DataFrame\n",
    "df_twitter.select(F.explode(F.col('hashtags'))).distinct().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 221,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|count(aux_col)|\n",
      "+--------------+\n",
      "|          1215|\n",
      "+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table.\n",
    "spark.sql(\"SELECT \\\n",
    "          count(aux_col) \\\n",
    "          FROM ( \\\n",
    "            SELECT \\\n",
    "            DISTINCT(EXPLODE(hashtags)) as aux_col \\\n",
    "            FROM twitter \\\n",
    "          ) aux\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.6 Get top 5 hashtags"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 223,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('طبق_القدرات_للثانويه_ياريس', 385),\n",
       " ('Corona', 319),\n",
       " ('OilPrice', 251),\n",
       " ('COVID19', 125),\n",
       " ('corona', 123)]"
      ]
     },
     "execution_count": 223,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD (4 points)\n",
    "rdd_twitter.flatMap(lambda x: x[2]).map(lambda x: (x,1)).reduceByKey(lambda x,y: x+y).sortBy(lambda x: x[1],ascending=False).take(5)\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 231,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(hashtags='طبق_القدرات_للثانويه_ياريس', count=385),\n",
       " Row(hashtags='Corona', count=319),\n",
       " Row(hashtags='OilPrice', count=251),\n",
       " Row(hashtags='COVID19', count=125),\n",
       " Row(hashtags='corona', count=123)]"
      ]
     },
     "execution_count": 231,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using DataFrame (2 points)\n",
    "df_twitter.select(F.explode(F.col('hashtags')).alias('hashtags')).groupBy('hashtags').count().sort(F.col('count').desc()).head(5)            "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 239,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------------+\n",
      "|       expl_hashtags|count_hasthag|\n",
      "+--------------------+-------------+\n",
      "|طبق_القدرات_للثان...|          385|\n",
      "|              Corona|          319|\n",
      "|            OilPrice|          251|\n",
      "|             COVID19|          125|\n",
      "|              corona|          123|\n",
      "+--------------------+-------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table. (2 points)\n",
    "spark.sql(\"SELECT \\\n",
    "          expl_hashtags, \\\n",
    "          COUNT(expl_hashtags) as count_hasthag  \\\n",
    "          FROM (\\\n",
    "            SELECT EXPLODE(hashtags) as expl_hashtags \\\n",
    "            FROM twitter \\\n",
    "          ) \\\n",
    "          GROUP BY expl_hashtags \\\n",
    "          ORDER BY count_hasthag DESC\").show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.6 Get total number of tweets which are retweeted more than 100 times"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 243,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[created_at: string, favorite_count: bigint, hashtags: array<string>, id: string, in_reply_to_status_id: string, in_reply_to_user_id_str: string, location: string, reply_count: bigint, retweet_count: bigint, source: string, user_id_str: string, user_followers_count: bigint, user_friends_count: bigint, user_created_at: string, extracted_source: string]"
      ]
     },
     "execution_count": 243,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_twitter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 246,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "15753"
      ]
     },
     "execution_count": 246,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD\n",
    "rdd_twitter.filter(lambda x: x[8]>100).count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 248,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "15753"
      ]
     },
     "execution_count": 248,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using DataFrame\n",
    "df_twitter.filter(F.col(\"retweet_count\")>100).count()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 249,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count_id|\n",
      "+--------+\n",
      "|   15753|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table.\n",
    "spark.sql(\"SELECT \\\n",
    "            COUNT(id) as count_id \\\n",
    "            FROM twitter \\\n",
    "            WHERE retweet_count>100\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.8 Get top 3 most retweeted tweets per country (8 points)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 279,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('India',\n",
       "  [['1252332114948874240', 9988],\n",
       "   ['1252252336921206787', 9976],\n",
       "   ['1252254519116746754', 9973]]),\n",
       " ('Pakistan',\n",
       "  [['1252334264248606720', 9988],\n",
       "   ['1252251912084357121', 9975],\n",
       "   ['1252252126694309888', 9973]]),\n",
       " ('USA',\n",
       "  [['1252331777806524416', 9994],\n",
       "   ['1252254239805579264', 9987],\n",
       "   ['1252335464750735362', 9982]]),\n",
       " ('Italy',\n",
       "  [['1252252106750377996', 9994],\n",
       "   ['1252251206027816960', 9984],\n",
       "   ['1252330500670337024', 9971]]),\n",
       " ('Canada',\n",
       "  [['1252335430323888128', 9997],\n",
       "   ['1252254877939531776', 9992],\n",
       "   ['1252252082825986051', 9987]]),\n",
       " ('China',\n",
       "  [['1252335780707684352', 9998],\n",
       "   ['1252253596516843520', 9993],\n",
       "   ['1252255562525560832', 9984]]),\n",
       " ('Chile',\n",
       "  [['1252253612140490759', 9988],\n",
       "   ['1252334891951427585', 9984],\n",
       "   ['1252253710182481920', 9978]]),\n",
       " ('UK',\n",
       "  [['1252333018578145280', 9991],\n",
       "   ['1252252091822870529', 9989],\n",
       "   ['1252254043973603329', 9985]]),\n",
       " ('Mexico',\n",
       "  [['1252253843145912320', 9998],\n",
       "   ['1252255209776189442', 9994],\n",
       "   ['1252252016006422533', 9971]]),\n",
       " ('Spain',\n",
       "  [['1252335445876367361', 9992],\n",
       "   ['1252334839094599681', 9981],\n",
       "   ['1252254696112300032', 9969]]),\n",
       " ('Germany',\n",
       "  [['1252334028092399622', 9999],\n",
       "   ['1252330902325248000', 9997],\n",
       "   ['1252252295510855682', 9990]])]"
      ]
     },
     "execution_count": 279,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD # return Get top 3 most retweeted tweets per country\n",
    "\n",
    "# optimize the code above using groupByKey and mapValues without numpy and sort\n",
    "rdd_twitter.map(lambda x: (x[6],[x[3],x[8]])).groupByKey().mapValues(lambda x: sorted(list(x),key=lambda x: x[1],reverse=True)[:3]).collect()\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 264,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+--------+-------------+-----------+\n",
      "|                 id|location|retweet_count|rt_loc_rank|\n",
      "+-------------------+--------+-------------+-----------+\n",
      "|1252335430323888128|  Canada|         9997|          1|\n",
      "|1252254877939531776|  Canada|         9992|          2|\n",
      "|1252252082825986051|  Canada|         9987|          3|\n",
      "|1252253612140490759|   Chile|         9988|          1|\n",
      "|1252334891951427585|   Chile|         9984|          2|\n",
      "|1252253710182481920|   Chile|         9978|          3|\n",
      "|1252335780707684352|   China|         9998|          1|\n",
      "|1252253596516843520|   China|         9993|          2|\n",
      "|1252255562525560832|   China|         9984|          3|\n",
      "|1252334028092399622| Germany|         9999|          1|\n",
      "|1252330902325248000| Germany|         9997|          2|\n",
      "|1252252295510855682| Germany|         9990|          3|\n",
      "|1252332114948874240|   India|         9988|          1|\n",
      "|1252252336921206787|   India|         9976|          2|\n",
      "|1252254519116746754|   India|         9973|          3|\n",
      "|1252252106750377996|   Italy|         9994|          1|\n",
      "|1252251206027816960|   Italy|         9984|          2|\n",
      "|1252330500670337024|   Italy|         9971|          3|\n",
      "|1252253843145912320|  Mexico|         9998|          1|\n",
      "|1252255209776189442|  Mexico|         9994|          2|\n",
      "+-------------------+--------+-------------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using DataFrame\n",
    "from pyspark.sql.window import Window\n",
    "df_twitter.select('id', 'location', 'retweet_count').withColumn('rt_loc_rank', F.dense_rank().over(Window.partitionBy('location').orderBy(F.desc('retweet_count')))).filter(F.col('rt_loc_rank')<=3).show()\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 250,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-------------+-----------+\n",
      "|location|                 id|retweet_count|rt_loc_rank|\n",
      "+--------+-------------------+-------------+-----------+\n",
      "|  Canada|1252335430323888128|         9997|          1|\n",
      "|  Canada|1252254877939531776|         9992|          2|\n",
      "|  Canada|1252252082825986051|         9987|          3|\n",
      "|   Chile|1252253612140490759|         9988|          1|\n",
      "|   Chile|1252334891951427585|         9984|          2|\n",
      "|   Chile|1252253710182481920|         9978|          3|\n",
      "|   China|1252335780707684352|         9998|          1|\n",
      "|   China|1252253596516843520|         9993|          2|\n",
      "|   China|1252255562525560832|         9984|          3|\n",
      "| Germany|1252334028092399622|         9999|          1|\n",
      "| Germany|1252330902325248000|         9997|          2|\n",
      "| Germany|1252252295510855682|         9990|          3|\n",
      "|   India|1252332114948874240|         9988|          1|\n",
      "|   India|1252252336921206787|         9976|          2|\n",
      "|   India|1252254519116746754|         9973|          3|\n",
      "|   Italy|1252252106750377996|         9994|          1|\n",
      "|   Italy|1252251206027816960|         9984|          2|\n",
      "|   Italy|1252330500670337024|         9971|          3|\n",
      "|  Mexico|1252253843145912320|         9998|          1|\n",
      "|  Mexico|1252255209776189442|         9994|          2|\n",
      "+--------+-------------------+-------------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table.\n",
    "spark.sql(\"SELECT * \\\n",
    "          FROM (\\\n",
    "            SELECT \\\n",
    "            location,\\\n",
    "            id,\\\n",
    "            retweet_count,\\\n",
    "            ROW_NUMBER() \\\n",
    "            OVER (PARTITION BY location ORDER BY retweet_count DESC) as rt_loc_rank \\\n",
    "            FROM twitter\\\n",
    "          ) aux_table \\\n",
    "          WHERE rt_loc_rank <= 3\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2.9 Total number of tweets per country"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 286,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "defaultdict(int,\n",
       "            {'India': 1480,\n",
       "             'UK': 1376,\n",
       "             'Spain': 1464,\n",
       "             'USA': 1539,\n",
       "             'Canada': 1441,\n",
       "             'China': 1457,\n",
       "             'Pakistan': 1470,\n",
       "             'Germany': 1426,\n",
       "             'Mexico': 1409,\n",
       "             'Chile': 1410,\n",
       "             'Italy': 1422})"
      ]
     },
     "execution_count": 286,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Using RDD (3 points)\n",
    "rdd_twitter.map(lambda x: x[6]).countByValue()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 287,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-----+\n",
      "|location|count|\n",
      "+--------+-----+\n",
      "|     USA| 1539|\n",
      "|   India| 1480|\n",
      "|Pakistan| 1470|\n",
      "|   Spain| 1464|\n",
      "|   China| 1457|\n",
      "|  Canada| 1441|\n",
      "| Germany| 1426|\n",
      "|   Italy| 1422|\n",
      "|   Chile| 1410|\n",
      "|  Mexico| 1409|\n",
      "|      UK| 1376|\n",
      "+--------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using DataFrame (2 points)\n",
    "df_twitter.select('location').groupBy('location').count().orderBy(F.col('count').desc()).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 288,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------------------+\n",
      "|location|tweets_per_country|\n",
      "+--------+------------------+\n",
      "| Germany|              1426|\n",
      "|   India|              1480|\n",
      "|   China|              1457|\n",
      "|   Chile|              1410|\n",
      "|   Italy|              1422|\n",
      "|   Spain|              1464|\n",
      "|     USA|              1539|\n",
      "|  Mexico|              1409|\n",
      "|      UK|              1376|\n",
      "|  Canada|              1441|\n",
      "|Pakistan|              1470|\n",
      "+--------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Using spark.sql and the temporay table. (1 point)\n",
    "spark.sql(\"SELECT \\\n",
    "          location,\\\n",
    "          COUNT(location) as tweets_per_country \\\n",
    "          FROM twitter \\\n",
    "          GROUP BY location\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Save Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1 save the data such that you have seperate folder per country (2 points)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 289,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Using DataFrame\n",
    "df_twitter.write.partitionBy(\"location\").format(\"json\").save(\"corona_tweet_by_loc.json\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.2 Save the data as parquet files (1 points)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 290,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Using DataFrame\n",
    "df_twitter.write.format(\"parquet\").save(\"corona_tweet_out.parquet\")\n",
    "    "
   ]
  }
 ],
 "metadata": {
  "interpreter": {
   "hash": "4cd865de445bc63df39d026045225af818f47dbb17c8b4b10470a186decd8c1e"
  },
  "kernelspec": {
   "display_name": "Python 3.9.12 ('big_data')",
   "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.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}