Day #26 Analyzing_Tweets_PySpark_Batch_Analytics.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Analyzing Tweets_PySpark_Batch Analytics.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "toc_visible": true,
      "authorship_tag": "ABX9TyMB/NL2cjXYSFOies2EEj9n",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/skyprince999/100-Days-Of-ML/blob/master/Day%2026%23%20Analyzing_Tweets_PySpark_Batch_Analytics.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "sScqs9LISYBx",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from google.colab import drive\n",
        "drive.mount('/content/drive')"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Y7qPzWe5et5l",
        "colab_type": "text"
      },
      "source": [
        "### Install spark dependencies"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "S3PWSgrib82k",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "TaFsoCRnexGa",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!wget https://downloads.apache.org/spark/spark-2.4.6/spark-2.4.6-bin-hadoop2.6.tgz"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "JhxFAJCMe49f",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!tar xf spark-2.4.6-bin-hadoop2.6.tgz"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "sAyiyWkpfCTt",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!pip install -q findspark\n",
        "!pip install pyspark"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "nXJk6jID3fCS",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!wget http://apachemirror.wuchna.com/hadoop/common/stable/hadoop-3.2.1.tar.gz\n",
        "!tar xf hadoop-3.2.1.tar.gz"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dzhkBfKDfPxE",
        "colab_type": "text"
      },
      "source": [
        "### Setup required environment variables"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "FWgbujJ8fJ0a",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark-2.4.6-bin-hadoop2.6\"\n",
        "os.environ['HADOOP_HOME'] = '/content/hadoop-3.2.1'"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "SJ8ATKXGfgZz",
        "colab_type": "text"
      },
      "source": [
        "### Import required libraries"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ay2sy0_mfVJL",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark import SparkContext\n",
        "from pyspark.sql import SparkSession\n",
        "\n",
        "from pyspark.ml import Pipeline\n",
        "from pyspark.ml.classification import RandomForestClassifier\n",
        "from pyspark.ml.feature import IndexToString, StringIndexer, VectorIndexer, VectorAssembler\n",
        "from pyspark.ml.evaluation import MulticlassClassificationEvaluator\n",
        "\n",
        "from pyspark.sql.functions import isnan, when, count, col"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "LfVN8LK5lDMN",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.sql.functions import UserDefinedFunction\n",
        "from pyspark.sql.types import StringType\n",
        "from pyspark.sql.types import IntegerType\n",
        "from pyspark.sql.functions import *\n",
        "\n",
        "import json"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "J4xFiGyqpnOu",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import pandas as pd\n",
        "import numpy as np"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Gu1Y1TEufs1_",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Connect to the Spark server\n",
        "SparkContext.setSystemProperty('spark.executor.memory', '4g')\n",
        "\n",
        "spark = SparkSession.builder.appName(\"Covid-Tweet\").master(\"local[*]\").getOrCreate()\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ScoJ2yeJgqwv",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 50
        },
        "outputId": "5c3cc478-1542-4e07-ca7d-338a148f3dfa"
      },
      "source": [
        "%%time\n",
        "!mkdir SPARK-2020-01\n",
        "!cp drive/\"My Drive\"/\"SPARK+AI Hackathon 2020\"/SPARK-2020-05/*.*  SPARK-2020-01/\n",
        "\n",
        "!cp drive/\"My Drive\"/\"SPARK+AI Hackathon 2020\"/SPARK-2020-01/*.*  SPARK-2020-01/"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "CPU times: user 1.16 s, sys: 202 ms, total: 1.36 s\n",
            "Wall time: 8min 11s\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "TEq8_B4Bg6Kn",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# filename05 = os.listdir('SPARK-2020-05')\n",
        "# filename05 = [ 'SPARK-2020-05/' + f for f in filename05]\n",
        "# print(len(filename05))\n",
        "\n",
        "# filename01 = os.listdir('SPARK-2020-01')\n",
        "# filename01 = [ 'SPARK-2020-01/' + f for f in filename01]\n",
        "# print(len(filename01))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "EttInxlcf4sy",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 50
        },
        "outputId": "a91a459b-bf5e-4f6b-c8dd-29dc565f93af"
      },
      "source": [
        "%%time\n",
        "# Create data frame. spark.read.json >>> can also read from a directory \n",
        "json_file_path = 'SPARK-2020-01'\n",
        "df = spark.read.json(json_file_path)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "CPU times: user 234 ms, sys: 59.6 ms, total: 294 ms\n",
            "Wall time: 25min 8s\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "TaQlIHQ6jFRZ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# The inferred schema can be visualized using the printSchema() method\n",
        "#df.printSchema()\n",
        "\n",
        "#df.show(5)\n",
        "\n",
        "#df.dtypes\n",
        "\n",
        "#print((df.count(), len(df.columns)))\n",
        "\n",
        "# Creates a temporary view using the DataFrame\n",
        "#df.createOrReplaceTempView(\"tweets\")\n",
        "\n",
        "# SQL statements can be run by using the sql methods provided by spark\n",
        "#tweetsDF = spark.sql(\"SELECT * FROM tweets\")\n",
        "#tweetsDF.show(10)\n",
        "\n",
        "#df.describe('favorite_count').show()\n",
        "\n",
        "# The following code block gives a null value\n",
        "\n",
        "# from pyspark.sql import functions as F\n",
        "\n",
        "# df2 = spark.createDataFrame([('Tue May 05 16:14:52 +0000 2020',)], ['t'])\n",
        "# df2 = df2.withColumn('new_date', df2.t.substr(1,19))\n",
        "# df2 = df2.withColumn('dt', F.to_date(df2.new_date, 'EEE MMM dd HH:MM:SS'))\n",
        "# print(df2.show())\n",
        "\n",
        "# This gives a OOM error \n",
        "# result = df.select('*').toPandas()\n",
        "\n",
        "## Instead add limit(n)\n",
        "# result = df.select('*').limit(100).toPandas()\n",
        "\n",
        "# df = df.withColumn('Day', df.created_at.substr(9,10))\n",
        "# df = df.withColumn('Hour', df.created_at.substr(12,13))\n",
        "\n",
        "# df.show(10, truncate=False)\n",
        "\n",
        "# df.select('entities', 'full_text').show(100,truncate=False)\n",
        "\n",
        "# from pyspark.sql.functions import flatten\n",
        "# df.select(df.full_text,flatten(df.entities)).show(100, truncate=False)\n",
        "\n",
        "# df.select(\"full_text\" , \"user\").show(10, truncate=False)\n",
        "\n",
        "# # Creates a temporary view using the DataFrame\n",
        "# df.createOrReplaceTempView(\"tweets\")\n",
        "\n",
        "# # SQL statements can be run by using the sql methods provided by spark\n",
        "# tweetsDF = spark.sql(\"SELECT * FROM tweets\").limit(100)\n",
        "# #tweetsDF.show(10)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "k1QftOQLe36a",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 222
        },
        "outputId": "fc5b1459-ae60-4f17-9ed3-451a2f5c1c5a"
      },
      "source": [
        "df.show(5)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+------------+-----------+--------------------+------------------+--------------------+-----------------+--------------+---------+--------------------+----+-------------------+-------------------+-----------------------+---------------------+-------------------------+-------------------+-----------------------+---------------+----+-----+------------------+-------------+----------------+--------------------+-----------------------+-------------+---------+--------------------+--------------------+---------+--------------------+------------------+---------------------+--------------+\n",
            "|contributors|coordinates|          created_at|display_text_range|            entities|extended_entities|favorite_count|favorited|           full_text| geo|                 id|             id_str|in_reply_to_screen_name|in_reply_to_status_id|in_reply_to_status_id_str|in_reply_to_user_id|in_reply_to_user_id_str|is_quote_status|lang|place|possibly_sensitive|quoted_status|quoted_status_id|quoted_status_id_str|quoted_status_permalink|retweet_count|retweeted|    retweeted_status|              source|truncated|                user|withheld_copyright|withheld_in_countries|withheld_scope|\n",
            "+------------+-----------+--------------------+------------------+--------------------+-----------------+--------------+---------+--------------------+----+-------------------+-------------------+-----------------------+---------------------+-------------------------+-------------------+-----------------------+---------------+----+-----+------------------+-------------+----------------+--------------------+-----------------------+-------------+---------+--------------------+--------------------+---------+--------------------+------------------+---------------------+--------------+\n",
            "|        null|       null|Thu Jan 30 20:00:...|           [0, 99]|[[],, [], [[a.msn...|             null|             0|    false|Coronavirus inter...|null|1222972750039408640|1222972750039408640|                   null|                 null|                     null|               null|                   null|          false|  en| null|              true|         null|            null|                null|                   null|            0|    false|                null|<a href=\"http://t...|    false|[false, Wed Apr 1...|              null|                 null|          null|\n",
            "|        null|       null|Thu Jan 30 20:00:...|          [0, 140]|[[[[48, 60], coro...|             null|             0|    false|RT @evankirstel: ...|null|1222972750068887552|1222972750068887552|                   null|                 null|                     null|               null|                   null|          false|  en| null|             false|         null|            null|                null|                   null|            9|    false|[,, Thu Jan 30 19...|<a href=\"http://t...|    false|[false, Mon Apr 1...|              null|                 null|          null|\n",
            "|        null|       null|Thu Jan 30 20:00:...|          [0, 140]|[[],, [], [[lapto...|             null|             0|    false|RT @CapriLaptop: ...|null|1222972750144376832|1222972750144376832|                   null|                 null|                     null|               null|                   null|          false|  en| null|             false|         null|            null|                null|                   null|            1|    false|[,, Thu Jan 30 19...|<a href=\"http://t...|    false|[false, Sat Apr 1...|              null|                 null|          null|\n",
            "|        null|       null|Thu Jan 30 20:00:...|          [0, 140]|[[],, [], [], [[2...|             null|             0|    false|RT @ElliottKayboo...|null|1222972750161039360|1222972750161039360|                   null|                 null|                     null|               null|                   null|          false|  en| null|              null|         null|            null|                null|                   null|        30647|    false|[,, Wed Jan 29 23...|<a href=\"http://t...|    false|[false, Fri Apr 2...|              null|                 null|          null|\n",
            "|        null|       null|Thu Jan 30 20:00:...|          [0, 116]|[[],, [], [[wapo....|             null|             0|    false|RT @PostWorld: In...|null|1222972750186156035|1222972750186156035|                   null|                 null|                     null|               null|                   null|          false|  en| null|             false|         null|            null|                null|                   null|          168|    false|[,, Thu Jan 30 09...|<a href=\"http://t...|    false|[false, Sat Apr 0...|              null|                 null|          null|\n",
            "+------------+-----------+--------------------+------------------+--------------------+-----------------+--------------+---------+--------------------+----+-------------------+-------------------+-----------------------+---------------------+-------------------------+-------------------+-----------------------+---------------+----+-----+------------------+-------------+----------------+--------------------+-----------------------+-------------+---------+--------------------+--------------------+---------+--------------------+------------------+---------------------+--------------+\n",
            "only showing top 5 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3T28t9xdjOWH",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "e00fe8ed-6421-445d-a4e9-34d02ea38415"
      },
      "source": [
        "df.count()"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "17165069"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 17
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "7IqrXdzPlUaV",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def user(text):\n",
        "  return text['screen_name']\n",
        "\n",
        "def hashtag(text):\n",
        "  return [h[1] for h in text[0]]\n",
        "  #return text[0]\n",
        "\n",
        "def user_mention(text):\n",
        "  return [h[-1] for h in text[-1]]"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4BD5lSjjm4h7",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.sql.types import ArrayType \n",
        "\n",
        "# Define your function\n",
        "getSN = UserDefinedFunction(lambda x: user(x), StringType())\n",
        "getHash = UserDefinedFunction(lambda x: hashtag(x), ArrayType(StringType()))\n",
        "getUM = UserDefinedFunction(lambda x: user_mention(x), ArrayType(StringType()))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "3hF50OqVDOkf",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "user = df.withColumn('user', getSN(col('user'))) #.show(5)  #<<- This works "
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "KxX1faxpDTFO",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "user = user.withColumn('entities', explode(array('entities'))) #.show(5)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "PHanrXWhHgpK",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "user = user.withColumn('hashtags', getHash('entities')) #.show(50)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "rO8hGNmbMIRW",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "user = user.withColumn('user_mention', getUM('entities'))"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "RK9-mgNTBDEH",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 1000
        },
        "outputId": "5617aa7d-73ef-4ee1-eebe-5201ea94a73f"
      },
      "source": [
        "user.select('user', 'hashtags', 'user_mention', 'full_text').show(50)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+---------------+--------------------+--------------------+--------------------+\n",
            "|           user|            hashtags|        user_mention|           full_text|\n",
            "+---------------+--------------------+--------------------+--------------------+\n",
            "|   ivaforgotton|                  []|                  []|Coronavirus inter...|\n",
            "|     iivanov482|[coronavirus, Cor...|       [evankirstel]|RT @evankirstel: ...|\n",
            "|websterelizabet|                  []|       [CapriLaptop]|RT @CapriLaptop: ...|\n",
            "|newagebitchface|                  []|   [ElliottKaybooks]|RT @ElliottKayboo...|\n",
            "|     rtsang0718|                  []|         [PostWorld]|RT @PostWorld: In...|\n",
            "|        mvintel|                  []|       [ali20waheed]|@ali20waheed stil...|\n",
            "|     ramzyyyyy_|                  []|                  []|CNN: McGill Unive...|\n",
            "|        linod99|                  []|         [emtpyvase]|RT @emtpyvase: so...|\n",
            "|   Iamathelstan|                  []|           [FLUMEED]|RT @FLUMEED: Aucu...|\n",
            "|   MableKipenda|                  []|       [BBCBreaking]|RT @BBCBreaking: ...|\n",
            "|      YoTrekkie|                  []|                  []|https://t.co/bXqg...|\n",
            "|      sherifink|                  []|[WHO, DrTedros, WHO]|@WHO @DrTedros Th...|\n",
            "|     milesboard|  [AI, deeplearning]|                  []|In coronavirus re...|\n",
            "|      FOX42KPTM|                  []|                  []|The new virus has...|\n",
            "|    FEEZYDoesIT|       [coronavirus]|                  []|Coronavirus decla...|\n",
            "|    tcsnoticias|          [Economía]|                  []|#Economía\n",
            "EE. UU....|\n",
            "|     ElSoldeMzt|[Mundo, Coronavir...|[RuizEspinoza74, ...|#Mundo #Coronavir...|\n",
            "| elpoliticonews|[EEUU, Japón, Chi...|                  []|#EEUU y #Japón ev...|\n",
            "|    indypremium|                  []|                  []|Opinion: The coro...|\n",
            "|   olhardigital|       [Coronavírus]|                  []|[EXCLUSIVO] #Coro...|\n",
            "|   Observateurs|[Chine, coronavir...|                  []|Ces derniers jour...|\n",
            "|   JoshanSubash|                  []|               [WHO]|@WHO 😂🤣🤣🤣🤣🤣...|\n",
            "|     world_ecot|                  []|       [ChuBailiang]|RT @ChuBailiang: ...|\n",
            "|      AmberGoth|                  []|[AyazEvrende, Dan...|@AyazEvrende @Dan...|\n",
            "|AVTechnologyMag|[CoronavirusOutbr...|          [ISE_Show]|.@ISE_Show has re...|\n",
            "|   Just_Koketso|                  []|           [Lemonka]|@Lemonka China ma...|\n",
            "|  tartanpirate1|       [coronavirus]|   [DrEricDing, WHO]|RT @DrEricDing: B...|\n",
            "|      canalplus|[Coronavirus, Bro...|    [BertrandUsclat]|Quand un hypocond...|\n",
            "|         TN21sv|              [Nota]|                  []|#Nota |🙆‍♂️😷 At...|\n",
            "|  DAWNCATHERINE|                  []|                  []|https://t.co/Fdju...|\n",
            "|      EFEMexico|       [coronavirus]|                  []|Primer caso de co...|\n",
            "| wakeupengland1|                  []|[OSRSLukeee, BNOD...|RT @OSRSLukeee: @...|\n",
            "|   americaasia_|[SanFrancisco, Al...|                  []|#SanFrancisco es ...|\n",
            "|AnaPintoCoelho1|          [2019nCoV]|               [WHO]|RT @WHO: LIVE: Pr...|\n",
            "|     AbanteNews|                  []|                  []|ICYMI: Kahit may ...|\n",
            "|  puntoporpunto|                  []|                  []|CORONAVIRUS manti...|\n",
            "|         mwas_2|                  []|            [cnnbrk]|RT @cnnbrk: Six t...|\n",
            "|     Diego1972_|                  []|[WHO, DrTedros, W...|RT @WHO: @DrTedro...|\n",
            "|  tootsierollex|                  []|       [M4ST3RSP4RK]|RT @M4ST3RSP4RK: ...|\n",
            "|     EddyWarman|[ÚltimaHora, públ...|               [WHO]|#ÚltimaHora \n",
            "\n",
            "La ...|\n",
            "|LeandroMachuca7|             [AHORA]|               [C5N]|RT @C5N: #AHORA: ...|\n",
            "| ClausDrastrup1|          [2019nCoV]|               [WHO]|RT @WHO: LIVE: Pr...|\n",
            "|Pasquale_paki07|                  []|   [ShooterHatesYou]|RT @ShooterHatesY...|\n",
            "|  QuinKirweiner|                  []|       [sjropequeen]|RT @sjropequeen: ...|\n",
            "|    NDtitulares|                  []|                  []|Rusia anunció el ...|\n",
            "|    maka_black_|                  []|          [m_ebrard]|@m_ebrard El prob...|\n",
            "|   gucxi__gucci|                  []|                  []|Madre mía conejo ...|\n",
            "|   SinEmbargoMX|                  []|                  []|Estados Unidos an...|\n",
            "|jorgecabral1201|                  []|    [infobaeamerica]|RT @infobaeameric...|\n",
            "|   NACCHOalerts|                  []|[WHO, DrTedros, W...|RT @WHO: @DrTedro...|\n",
            "+---------------+--------------------+--------------------+--------------------+\n",
            "only showing top 50 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "7X5xY1L4OmKd",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 84
        },
        "outputId": "af992a48-ae46-4dfd-8655-88495640fc68"
      },
      "source": [
        "query = user.select('user', 'hashtags', 'user_mention', 'full_text')\n",
        "query.dtypes # csv can't save a csv file with columns in array<strin> format"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[('user', 'string'),\n",
              " ('hashtags', 'array<string>'),\n",
              " ('user_mention', 'array<string>'),\n",
              " ('full_text', 'string')]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 25
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "28caLQogRozP",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 84
        },
        "outputId": "61ebb121-8e4f-4971-d255-921061fd9469"
      },
      "source": [
        "sparse_format_udf = udf(lambda x: ','.join([str(elem) for elem in x], StringType()))\n",
        "\n",
        "query = query.withColumn('hashtags', sparse_format_udf(col('hashtags')))\n",
        "query = query.withColumn('user_mention', sparse_format_udf(col('user_mention')))\n",
        "query.dtypes"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[('user', 'string'),\n",
              " ('hashtags', 'string'),\n",
              " ('user_mention', 'string'),\n",
              " ('full_text', 'string')]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 26
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "x3W5C79UPS5e",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 34
        },
        "outputId": "2f6572d6-5c78-474a-c169-8440b5c9983b"
      },
      "source": [
        "type(query)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "pyspark.sql.dataframe.DataFrame"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 29
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "6ybXcmEHLXKp",
        "colab_type": "code",
        "colab": {
          "base_uri": "https://localhost:8080/",
          "height": 54
        },
        "outputId": "d9011cb9-2d80-4ec9-9567-1d1e4d295bbd"
      },
      "source": [
        "'''\n",
        "The following statements are causing an error \n",
        ">>>\n",
        "raise Py4JError(\n",
        "\n",
        "An error occurred while calling o278.collectToPython\n",
        "\n",
        "TypeError: join() takes exactly one argument (2 given)\n",
        "'''\n",
        "# query.limit(1000).toPandas().to_excel('Output.xlsx')\n",
        "\n",
        "# # Save file local folder, delimiter by default is ,\n",
        "# query.coalesce(1).write.format('csv').option('header',\"True\").mode('overwrite').option('sep',',').save('output')\n",
        "\n",
        "# query.write.format(\"parquet\").save(\"jan2020.parquet\")"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "'\\nThe following statements are causing an error \\n>>>\\nraise Py4JError(\\n\\nAn error occurred while calling o278.collectToPython\\n\\nTypeError: join() takes exactly one argument (2 given)\\n'"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 28
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "eqyRsH_fFO_K",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        ""
      ],
      "execution_count": null,
      "outputs": []
    }
  ]
}