01b Comments User Join.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *\n",
    "import time\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "text_file = sc.textFile(\"gs://stackoverflow-dataset-677/Comments.xml\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['<?xml version=\"1.0\" encoding=\"utf-8\"?>',\n",
       " '<comments>',\n",
       " '  <row Id=\"13\" PostId=\"23\" Score=\"0\" Text=\"Using /opt helps me keep track of the applications I\\'ve installed myself.\" CreationDate=\"2010-07-28T19:36:59.773\" UserId=\"10\" ContentLicense=\"CC BY-SA 2.5\" />']"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "text_file.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "filteredRDD = text_file.filter(lambda x: x.startswith(\"  <row \"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['  <row Id=\"13\" PostId=\"23\" Score=\"0\" Text=\"Using /opt helps me keep track of the applications I\\'ve installed myself.\" CreationDate=\"2010-07-28T19:36:59.773\" UserId=\"10\" ContentLicense=\"CC BY-SA 2.5\" />']"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "filteredRDD.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "cleanedRDD = filteredRDD.map(lambda x: x.lstrip(\"  \"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['<row Id=\"13\" PostId=\"23\" Score=\"0\" Text=\"Using /opt helps me keep track of the applications I\\'ve installed myself.\" CreationDate=\"2010-07-28T19:36:59.773\" UserId=\"10\" ContentLicense=\"CC BY-SA 2.5\" />']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "cleanedRDD.take(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {},
   "outputs": [],
   "source": [
    "import xml.etree.ElementTree as ET\n",
    "\n",
    "def parse_xml(rdd):\n",
    "    \"\"\"\n",
    "    Read the xml string from rdd, parse and extract the elements,\n",
    "    then return a list of list.\n",
    "    \"\"\"\n",
    "    root = ET.fromstring(rdd)\n",
    "    rec = []\n",
    "    \n",
    "    if \"PostId\" in root.attrib:\n",
    "        rec.append(int(root.attrib['PostId']))\n",
    "    else:\n",
    "        rec.append(0)\n",
    "\n",
    "    \n",
    "    if \"Score\" in root.attrib:\n",
    "        rec.append(int(root.attrib['Score']))\n",
    "    else:\n",
    "        rec.append(0)\n",
    "\n",
    "    \n",
    "    if \"Text\" in root.attrib:\n",
    "        rec.append(root.attrib['Text'])\n",
    "    else:\n",
    "        rec.append(\"N/A\")\n",
    "    \n",
    "    if \"CreationDate\" in root.attrib:\n",
    "        rec.append(root.attrib['CreationDate'])\n",
    "    else:\n",
    "        rec.append(\"N/A\")\n",
    "\n",
    "    if \"UserId\" in root.attrib:\n",
    "        rec.append(int(root.attrib['UserId']))\n",
    "    else:\n",
    "        rec.append(0)\n",
    "    return rec"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [],
   "source": [
    "records_rdd = cleanedRDD.map(lambda x : parse_xml(x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [],
   "source": [
    "comments_data = [\"postId\",\"score\",\"text\",\"creationDate\",\"userId\"]\n",
    "comments_df = records_rdd.toDF(comments_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- postId: long (nullable = true)\n",
      " |-- score: long (nullable = true)\n",
      " |-- text: string (nullable = true)\n",
      " |-- creationDate: string (nullable = true)\n",
      " |-- userId: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "comments_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+--------------------+--------------------+------+\n",
      "|postId|score|                text|        creationDate|userId|\n",
      "+------+-----+--------------------+--------------------+------+\n",
      "|    23|    0|Using /opt helps ...|2010-07-28T19:36:...|    10|\n",
      "|    18|    0|but popping in a ...|2010-07-28T19:38:...|    10|\n",
      "|    27|    0|That will revert ...|2010-07-28T19:39:...|    50|\n",
      "|    31|    0|I think you meant...|2010-07-28T19:41:...|    12|\n",
      "|    18|    0|@DLH apparently n...|2010-07-28T19:41:...|    63|\n",
      "|    12|    2|\"ssh -X <server> ...|2010-07-28T19:46:...|    96|\n",
      "|    12|    0|@Suppressingfire:...|2010-07-28T19:48:...|    10|\n",
      "|    50|    0|Can you please re...|2010-07-28T19:48:...|    56|\n",
      "|    27|    0|It probably shoul...|2010-07-28T19:49:...|     5|\n",
      "|    58|    0|Do you mean the c...|2010-07-28T19:50:...|     5|\n",
      "|    47|    0|Have you checked ...|2010-07-28T19:50:...|     4|\n",
      "|    47|    1|Might be related ...|2010-07-28T19:51:...|   104|\n",
      "|    58|    0|Do you use Gnome ...|2010-07-28T19:51:...|     4|\n",
      "|    60|    0|This causes data ...|2010-07-28T19:52:...|    66|\n",
      "|    18|    0|no the live CD do...|2010-07-28T19:53:...|     4|\n",
      "|    52|    0|Does this let the...|2010-07-28T19:55:...|    35|\n",
      "|    56|    2|LDAP and nfs are ...|2010-07-28T19:56:...|     4|\n",
      "|    10|    0|Can I use it on a...|2010-07-28T19:56:...|    27|\n",
      "|    70|    1|That's a good tip...|2010-07-28T19:56:...|    45|\n",
      "|    70|    0|That is probably ...|2010-07-28T19:58:...|    86|\n",
      "+------+-----+--------------------+--------------------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "comments_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|postId|score|\n",
      "+------+-----+\n",
      "|    23|    0|\n",
      "|    18|    0|\n",
      "|    27|    0|\n",
      "|    31|    0|\n",
      "|    18|    0|\n",
      "|    12|    2|\n",
      "|    12|    0|\n",
      "|    50|    0|\n",
      "|    27|    0|\n",
      "|    58|    0|\n",
      "|    47|    0|\n",
      "|    47|    1|\n",
      "|    58|    0|\n",
      "|    60|    0|\n",
      "|    18|    0|\n",
      "|    52|    0|\n",
      "|    56|    2|\n",
      "|    10|    0|\n",
      "|    70|    1|\n",
      "|    70|    0|\n",
      "+------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "comments_df.select(\"postId\",\"score\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [],
   "source": [
    "comments_df.createOrReplaceTempView(\"comments\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {},
   "outputs": [],
   "source": [
    "comments_sql_df = spark.sql(\"SELECT * FROM comments\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- postId: long (nullable = true)\n",
      " |-- score: long (nullable = true)\n",
      " |-- text: string (nullable = true)\n",
      " |-- creationDate: string (nullable = true)\n",
      " |-- userId: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "comments_sql_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+--------------------+--------------------+------+\n",
      "|postId|score|                text|        creationDate|userId|\n",
      "+------+-----+--------------------+--------------------+------+\n",
      "|    23|    0|Using /opt helps ...|2010-07-28T19:36:...|    10|\n",
      "|    18|    0|but popping in a ...|2010-07-28T19:38:...|    10|\n",
      "|    27|    0|That will revert ...|2010-07-28T19:39:...|    50|\n",
      "|    31|    0|I think you meant...|2010-07-28T19:41:...|    12|\n",
      "|    18|    0|@DLH apparently n...|2010-07-28T19:41:...|    63|\n",
      "|    12|    2|\"ssh -X <server> ...|2010-07-28T19:46:...|    96|\n",
      "|    12|    0|@Suppressingfire:...|2010-07-28T19:48:...|    10|\n",
      "|    50|    0|Can you please re...|2010-07-28T19:48:...|    56|\n",
      "|    27|    0|It probably shoul...|2010-07-28T19:49:...|     5|\n",
      "|    58|    0|Do you mean the c...|2010-07-28T19:50:...|     5|\n",
      "|    47|    0|Have you checked ...|2010-07-28T19:50:...|     4|\n",
      "|    47|    1|Might be related ...|2010-07-28T19:51:...|   104|\n",
      "|    58|    0|Do you use Gnome ...|2010-07-28T19:51:...|     4|\n",
      "|    60|    0|This causes data ...|2010-07-28T19:52:...|    66|\n",
      "|    18|    0|no the live CD do...|2010-07-28T19:53:...|     4|\n",
      "|    52|    0|Does this let the...|2010-07-28T19:55:...|    35|\n",
      "|    56|    2|LDAP and nfs are ...|2010-07-28T19:56:...|     4|\n",
      "|    10|    0|Can I use it on a...|2010-07-28T19:56:...|    27|\n",
      "|    70|    1|That's a good tip...|2010-07-28T19:56:...|    45|\n",
      "|    70|    0|That is probably ...|2010-07-28T19:58:...|    86|\n",
      "+------+-----+--------------------+--------------------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "comments_sql_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 106,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+--------------------+--------------------+------+\n",
      "|postId|score|                text|        creationDate|userId|\n",
      "+------+-----+--------------------+--------------------+------+\n",
      "|    23|    0|Using /opt helps ...|2010-07-28T19:36:...|    10|\n",
      "|    18|    0|but popping in a ...|2010-07-28T19:38:...|    10|\n",
      "|    12|    0|@Suppressingfire:...|2010-07-28T19:48:...|    10|\n",
      "|    77|    0|hmm that would be...|2010-07-28T20:05:...|    10|\n",
      "|    78|    0|oh really? hmm I ...|2010-07-28T20:35:...|    10|\n",
      "|   396|    1|Where can I get a...|2010-07-29T14:55:...|    10|\n",
      "|   405|    2|Huh I hadn't thou...|2010-07-29T17:10:...|    10|\n",
      "|   159|    0|I had some basic ...|2010-07-29T17:20:...|    10|\n",
      "|   331|    0|+5 Aw dang! Can't...|2010-07-29T17:54:...|    10|\n",
      "|   327|    5|     community wiki?|2010-07-29T17:56:...|    10|\n",
      "|   430|    0|@both: Agreed. Th...|2010-07-29T19:52:...|    10|\n",
      "|   797|    0|I don't think it'...|2010-08-04T18:12:...|    10|\n",
      "|   865|    4|Yeah I've been ve...|2010-08-04T22:53:...|    10|\n",
      "|  1062|    1|Hmm Lightspark lo...|2010-08-05T12:52:...|    10|\n",
      "|   822|   39|My own personal w...|2010-08-05T14:51:...|    10|\n",
      "|  1145|    2|@sepp2k: I think ...|2010-08-05T14:55:...|    10|\n",
      "|   822|   31|Is a fine wine an...|2010-08-05T15:15:...|    10|\n",
      "|   822|   11|Is 2001: A Space ...|2010-08-05T15:17:...|    10|\n",
      "|  2231|    1|Yes, I mentioned ...|2010-08-14T05:06:...|    10|\n",
      "| 15069|    1|The wiki.ubuntu.c...|2010-11-27T04:30:...|    10|\n",
      "+------+-----+--------------------+--------------------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM comments where userId = 10\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {},
   "outputs": [],
   "source": [
    "users_data = sc.textFile(\"gs://stackoverflow-dataset-677/users_out/*.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['1,Community', '2,Geoff Dalgas', '3,Jarrod Dixon']"
      ]
     },
     "execution_count": 108,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "users_data.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "metadata": {},
   "outputs": [],
   "source": [
    "def create_user(rdd):\n",
    "    rdd_split = rdd.split(\",\")\n",
    "    return [int(rdd_split[0]),rdd_split[1]]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {},
   "outputs": [],
   "source": [
    "users_rdd = users_data.map(lambda x: create_user(x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[[1, 'Community'], [2, 'Geoff Dalgas'], [3, 'Jarrod Dixon']]"
      ]
     },
     "execution_count": 111,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "users_rdd.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_data = [\"id\",\"username\"]\n",
    "user_df = users_rdd.toDF(user_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: long (nullable = true)\n",
      " |-- username: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+----------------+\n",
      "| id|        username|\n",
      "+---+----------------+\n",
      "|  1|       Community|\n",
      "|  2|    Geoff Dalgas|\n",
      "|  3|    Jarrod Dixon|\n",
      "|  4|      txwikinger|\n",
      "|  5|    Nathan Osman|\n",
      "|  6|          Emmett|\n",
      "|  7|           Helix|\n",
      "|  8| mechanical_meat|\n",
      "|  9|          Andrew|\n",
      "| 10|             DLH|\n",
      "| 11|   hannes.koller|\n",
      "| 12|   Michael Terry|\n",
      "| 13|   Keith Maurino|\n",
      "| 14|          Jweede|\n",
      "| 16|        Jeremy L|\n",
      "| 17|          tutuca|\n",
      "| 18|          excid3|\n",
      "| 20|   ParanoiaPuppy|\n",
      "| 21|            GeoD|\n",
      "| 22|Alan Featherston|\n",
      "+---+----------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_df.createOrReplaceTempView(\"users\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: long (nullable = true)\n",
      " |-- username: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+--------+\n",
      "| id|username|\n",
      "+---+--------+\n",
      "| 26| njpatel|\n",
      "+---+--------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM users where id=26\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+-------------------+------+-----+--------------------+--------------------+------+\n",
      "|  id|           username|postId|score|                text|        creationDate|userId|\n",
      "+----+-------------------+------+-----+--------------------+--------------------+------+\n",
      "| 964|Hendrik Brummermann|  4602|    0|I can confirm thi...|2010-10-13T21:37:...|   964|\n",
      "| 964|Hendrik Brummermann|118087|    0|They took it in d...|2012-04-28T06:17:...|   964|\n",
      "| 964|Hendrik Brummermann|638027|    0|I have the same i...|2015-08-03T13:26:...|   964|\n",
      "|1677|         eslambasha| 84949|    0|@fossfreedom i do...|2011-12-03T21:56:...|  1677|\n",
      "|1697|           Frxstrem| 16683|    0|@Marco, I know, I...|2010-12-08T22:36:...|  1697|\n",
      "|1697|           Frxstrem| 16784|    0|This seems to be ...|2010-12-09T19:05:...|  1697|\n",
      "|1697|           Frxstrem| 16886|    1|I only want to di...|2010-12-10T22:26:...|  1697|\n",
      "|1697|           Frxstrem| 16892|    1|This is not an ac...|2010-12-10T22:28:...|  1697|\n",
      "|1697|           Frxstrem| 16988|    0|Have you tried bu...|2010-12-11T19:22:...|  1697|\n",
      "|1697|           Frxstrem| 17471|    0|   @Stefano fixed it|2010-12-14T23:14:...|  1697|\n",
      "|1697|           Frxstrem| 17892|    0|My guess is that ...|2010-12-17T13:50:...|  1697|\n",
      "|1697|           Frxstrem| 18014|    0|-1 It's too uncle...|2010-12-18T17:53:...|  1697|\n",
      "|1697|           Frxstrem| 18273|    0|You did replace `...|2010-12-22T17:48:...|  1697|\n",
      "|1697|           Frxstrem| 67121|    0|Firstly, I have a...|2011-10-15T22:18:...|  1697|\n",
      "|1697|           Frxstrem|108944|    0|You should use `t...|2012-03-01T00:30:...|  1697|\n",
      "|1697|           Frxstrem|453415|    2|Daily builds can ...|2014-04-23T07:29:...|  1697|\n",
      "|1697|           Frxstrem|223442|    0|@user2662639 Simp...|2015-08-26T16:36:...|  1697|\n",
      "|1697|           Frxstrem|223442|    0|@user2662639 (I t...|2015-08-26T16:37:...|  1697|\n",
      "|1697|           Frxstrem| 17650|    2|@Fiksdal I don't ...|2016-03-25T12:21:...|  1697|\n",
      "|1697|           Frxstrem|899129|    0|@DavidFoerster Th...|2017-04-01T13:36:...|  1697|\n",
      "+----+-------------------+------+-----+--------------------+--------------------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "comments_users_sql_df = spark.sql(\"SELECT * FROM users u JOIN comments c ON u.id = c.UserId\")\n",
    "comments_users_sql_df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_data = [\"id\",\"username\"]\n",
    "user_df = users_rdd.toDF(user_data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {},
   "outputs": [],
   "source": [
    "comments_users_sql_df.createOrReplaceTempView(\"comments_users\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- id: long (nullable = true)\n",
      " |-- username: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "python",
   "name": "pyspark"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}