Colab_SocialMedia_Example_Spark.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Colab-SocialMedia-Example-Spark.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "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/kiat/BigDataAnalytics/blob/master/Spark-Example-Social-Media/Colab_SocialMedia_Example_Spark.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "LmxLySb20jtX"
      },
      "source": [
        "# Spark installation on Colab\n",
        "\n",
        "!pip install --ignore-installed -q pyspark==3.0.1"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Nor9olN20p0M"
      },
      "source": [
        "# Set JAVA_HOME and SPARK_HOME\n",
        "import os\n",
        "# os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-11-openjdk-amd64\"\n",
        "# os.environ[\"SPARK_HOME\"] = \"spark-3.0.1-bin-hadoop3.2\"\n",
        "\n",
        "# import findspark\n",
        "# findspark.init(\"spark-3.0.1-bin-hadoop3.2\")# SPARK_HOME\n",
        "\n",
        "\n",
        "import sys\n",
        "import requests\n",
        "from operator import add\n",
        "\n",
        "from pyspark import SparkConf,SparkContext\n",
        "from pyspark.streaming import StreamingContext\n",
        "\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql import SQLContext\n",
        "\n",
        "from pyspark.sql.types import *\n",
        "from pyspark.sql import functions as func\n",
        "from pyspark.sql.functions import *\n",
        "\n",
        "spark = SparkSession.builder.master(\"local[*]\").getOrCreate()\n",
        "sc = SparkContext.getOrCreate()\n",
        "sqlContext = SQLContext(sc)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "XL3CjM3406y4"
      },
      "source": [
        ""
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2bNmkjaMYk7r"
      },
      "source": [
        "# Download the data\n",
        "! wget -q https://s3.amazonaws.com/metcs777/person_knows_person.csv\n",
        "! wget -q https://s3.amazonaws.com/metcs777/person_likes_post.csv\n",
        "! wget -q https://s3.amazonaws.com/metcs777/post_hasCreator_person.csv\n",
        "! wget -q https://s3.amazonaws.com/metcs777/comment_hasCreator_person.csv\n",
        "! wget -q https://s3.amazonaws.com/metcs777/comment_replyOf_post.csv\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Q0qB4iql-uaK",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "25d30730-1da3-4224-8b2c-a4537798e9da"
      },
      "source": [
        "! ls -la "
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "total 33572\n",
            "drwxr-xr-x  1 root root     4096 Nov 11 21:26 .\n",
            "drwxr-xr-x  1 root root     4096 Nov 11 21:23 ..\n",
            "-rw-r--r--  1 root root  7398677 Dec 12  2018 comment_hasCreator_person.csv\n",
            "-rw-r--r--  1 root root  3392677 Dec 12  2018 comment_replyOf_post.csv\n",
            "drwxr-xr-x  1 root root     4096 Nov  6 17:30 .config\n",
            "-rw-r--r--  1 root root   232246 Dec 12  2018 person_knows_person.csv\n",
            "-rw-r--r--  1 root root 21372777 Aug 15 19:14 person_likes_post.csv\n",
            "-rw-r--r--  1 root root  1950946 Dec 12  2018 post_hasCreator_person.csv\n",
            "drwxr-xr-x  1 root root     4096 Nov  6 17:30 sample_data\n",
            "drwxr-xr-x 13 1000 1000     4096 Aug 28 09:22 spark-3.0.1-bin-hadoop3.2\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "iKSEjiPucGoa",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "06a5824e-5a50-4c7a-930b-b30eb4ff4f96"
      },
      "source": [
        "!pwd"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "/content\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "h5VEYx0pAgy8"
      },
      "source": [
        "# Set your file path here \n",
        "path=\"file:///content/\"\n",
        "\n",
        "# You have 5 files \n",
        "\n",
        "# comment_hasCreator_person.csv\n",
        "# comment_replyOf_post.csv\n",
        "# person_knows_person.csv\n",
        "# person_likes_post.csv\n",
        "# post_hasCreator_person.csv\n",
        "\n",
        "fileCommentHasCreator= path+\"comment_hasCreator_person.csv\"\n",
        "fileComment_replyOf_post = path+\"comment_replyOf_post.csv\"\n",
        "filePerson_knows_person= path+\"person_knows_person.csv\"\n",
        "filePerson_likes_post = path + \"person_likes_post.csv\"\n",
        "filePost_hasCreator_person = path + \"post_hasCreator_person.csv\""
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9DsYcQ5Yc_uN"
      },
      "source": [
        "#Create RDDs "
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "rz5WU1Iuccm0"
      },
      "source": [
        "def getRDD(mfile, sc):\n",
        "\t\tlines = sc.textFile(mfile)\n",
        "\t\t# Removing the Header Line \n",
        "\t\tlinesHeader = lines.first()\n",
        "\t\theader = sc.parallelize([linesHeader])\n",
        "\t\tlinesWithOutHeader = lines.subtract(header)\n",
        "\t\tmyRDD = linesWithOutHeader.map(lambda x: x.split('|'))\n",
        "\t\treturn myRDD"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jyaYcn99cSWV"
      },
      "source": [
        "\t# Create RDDs from files \n",
        "\tcommentHasCreator = getRDD(fileCommentHasCreator, sc)\n",
        "\tcomment_replyOf_post = getRDD(fileComment_replyOf_post, sc)\n",
        "\tperson_knows_person = getRDD(filePerson_knows_person, sc)\n",
        "\tperson_likes_post = getRDD(filePerson_likes_post, sc)\n",
        "\tpost_hasCreator_person = getRDD(filePost_hasCreator_person, sc)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "0gZJZ01KcUaf",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "07d6c66b-69ec-48b6-edae-9055e790349b"
      },
      "source": [
        "commentHasCreator.take(1)"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "execute_result",
          "data": {
            "text/plain": [
              "[['30', '457']]"
            ]
          },
          "metadata": {
            "tags": []
          },
          "execution_count": 11
        }
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "M91Aasfec8BL"
      },
      "source": [
        "# Create Dataframes"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "eHCWjR6rBDfE"
      },
      "source": [
        "commentHasCreatorDD = sqlContext.read.format('csv').options(header='true', inferSchema='true',  sep =\"|\").load(fileCommentHasCreator)\n",
        "comment_replyOf_postDD = sqlContext.read.format('csv').options(header='true', inferSchema='true',  sep =\"|\").load(fileComment_replyOf_post)\n",
        "person_knows_personDD = sqlContext.read.format('csv').options(header='true', inferSchema='true',  sep =\"|\").load(filePerson_knows_person)\n",
        "person_likes_postDD = sqlContext.read.format('csv').options(header='true', inferSchema='true',  sep =\"|\").load(filePerson_likes_post)\n",
        "post_hasCreator_personDD = sqlContext.read.format('csv').options(header='true', inferSchema='true',  sep =\"|\").load(filePost_hasCreator_person)\n"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "JTMIAdr1dmD2",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "b3cffec0-a60f-4fb7-de03-6e4a69199959"
      },
      "source": [
        "commentHasCreatorDD.show()\n",
        "comment_replyOf_postDD.show()\n",
        "person_knows_personDD.show()\n",
        "person_likes_postDD.show()\n",
        "post_hasCreator_personDD.show()\n"
      ],
      "execution_count": null,
      "outputs": [
        {
          "output_type": "stream",
          "text": [
            "+----------+---------+\n",
            "|Comment.id|Person.id|\n",
            "+----------+---------+\n",
            "|         0|       74|\n",
            "|        10|      832|\n",
            "|        20|      913|\n",
            "|        30|      457|\n",
            "|        40|      956|\n",
            "|        50|       41|\n",
            "|        60|      453|\n",
            "|        70|      832|\n",
            "|        80|        6|\n",
            "|        90|        6|\n",
            "|       100|      103|\n",
            "|       110|      547|\n",
            "|       120|      962|\n",
            "|       130|       99|\n",
            "|       140|      452|\n",
            "|       150|       99|\n",
            "|       160|        6|\n",
            "|       170|       48|\n",
            "|       180|      941|\n",
            "|       190|       40|\n",
            "+----------+---------+\n",
            "only showing top 20 rows\n",
            "\n",
            "+----------+-------+\n",
            "|Comment.id|Post.id|\n",
            "+----------+-------+\n",
            "|         0|      0|\n",
            "|        10|      0|\n",
            "|        30|      0|\n",
            "|        70|      0|\n",
            "|       100|     10|\n",
            "|       110|     10|\n",
            "|       140|     10|\n",
            "|       150|     10|\n",
            "|       180|     10|\n",
            "|       240|     20|\n",
            "|       250|     20|\n",
            "|       330|     20|\n",
            "|       340|     30|\n",
            "|       350|     30|\n",
            "|       420|     30|\n",
            "|       470|     40|\n",
            "|       480|     40|\n",
            "|       510|     40|\n",
            "|       560|     40|\n",
            "|       660|     50|\n",
            "+----------+-------+\n",
            "only showing top 20 rows\n",
            "\n",
            "+----------+----------+\n",
            "|Person.id0|Person.id1|\n",
            "+----------+----------+\n",
            "|        38|       956|\n",
            "|        38|       962|\n",
            "|        38|       941|\n",
            "|        38|        74|\n",
            "|        38|        36|\n",
            "|        38|        53|\n",
            "|        38|        48|\n",
            "|        38|        29|\n",
            "|        38|        46|\n",
            "|        38|        40|\n",
            "|        38|        60|\n",
            "|        38|        31|\n",
            "|        38|        41|\n",
            "|        38|         6|\n",
            "|        38|         4|\n",
            "|        38|       547|\n",
            "|        38|       832|\n",
            "|        38|       129|\n",
            "|        38|       915|\n",
            "|        38|       921|\n",
            "+----------+----------+\n",
            "only showing top 20 rows\n",
            "\n",
            "+---------+-------+-------------------+\n",
            "|Person.id|Post.id|       creationDate|\n",
            "+---------+-------+-------------------+\n",
            "|       74|      0|2012-10-15 05:13:41|\n",
            "|       36|      0|2012-10-18 10:51:39|\n",
            "|      417|     10|2012-11-18 11:14:47|\n",
            "|      415|     10|2012-11-20 06:30:22|\n",
            "|      109|     10|2012-11-19 11:03:52|\n",
            "|      119|     10|2012-11-16 09:07:14|\n",
            "|       70|     10|2012-11-19 04:53:43|\n",
            "|       47|     20|2012-10-23 03:32:14|\n",
            "|      378|     20|2012-10-25 01:03:21|\n",
            "|      643|     20|2012-10-18 11:45:28|\n",
            "|      592|     20|2012-10-18 06:02:36|\n",
            "|      576|     20|2012-10-21 02:23:03|\n",
            "|      606|     20|2012-10-24 01:03:00|\n",
            "|      724|     20|2012-10-23 00:17:38|\n",
            "|      672|     20|2012-10-24 04:32:12|\n",
            "|      711|     20|2012-10-22 04:38:54|\n",
            "|      915|     30|2012-06-22 09:10:46|\n",
            "|      921|     30|2012-06-19 03:22:50|\n",
            "|      507|     30|2012-06-23 07:15:14|\n",
            "|      501|     30|2012-06-23 00:34:27|\n",
            "+---------+-------+-------------------+\n",
            "only showing top 20 rows\n",
            "\n",
            "+-------+---------+\n",
            "|Post.id|Person.id|\n",
            "+-------+---------+\n",
            "|      0|       38|\n",
            "|     10|       38|\n",
            "|     20|       38|\n",
            "|     30|       38|\n",
            "|     40|       38|\n",
            "|     50|       38|\n",
            "|     60|       38|\n",
            "|     70|       38|\n",
            "|     80|       38|\n",
            "|     90|       38|\n",
            "|    100|       38|\n",
            "|    110|       38|\n",
            "|    120|       38|\n",
            "|    130|       38|\n",
            "|    140|       38|\n",
            "|    150|       38|\n",
            "|    160|       38|\n",
            "|    170|       38|\n",
            "|    180|       38|\n",
            "|    190|       38|\n",
            "+-------+---------+\n",
            "only showing top 20 rows\n",
            "\n"
          ],
          "name": "stdout"
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "XS4aiaZUEABC"
      },
      "source": [
        ""
      ],
      "execution_count": null,
      "outputs": []
    }
  ]
}