course2-big-data-analysis/c02w04-Demo.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Welcome to\n",
      "      ____              __\n",
      "     / __/__  ___ _____/ /__\n",
      "    _\\ \\/ _ \\/ _ `/ __/  '_/\n",
      "   /__ / .__/\\_,_/_/ /_/\\_\\   version 2.1.1\n",
      "      /_/\n",
      "\n",
      "Using Python version 2.7.12 (default, Nov 19 2016 06:48:10)\n",
      "SparkSession available as 'spark'.\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "execfile(os.path.join(os.environ[\"SPARK_HOME\"], 'python/pyspark/shell.py'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "sparkSession = SparkSession.builder.enableHiveSupport().master(\"local [2]\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "graphPath = \"/data/graphDFSample\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import explode, collect_list, size, col, row_number\n",
    "from pyspark.sql import Window\n",
    "\n",
    "reversedGraph = sparkSession.read.parquet(graphPath) \\\n",
    "    .withColumn(\"friend\", explode('friends')) \\\n",
    "    .groupBy(\"friend\") \\\n",
    "    .agg(collect_list(\"user\").alias(\"users\")) \\\n",
    "    .withColumn(\"users_size\", size(\"users\")) \\"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+----------+\n",
      "|friend|               users|users_size|\n",
      "+------+--------------------+----------+\n",
      "|   148|[65051219, 146311...|         4|\n",
      "|  5518|          [58573511]|         1|\n",
      "|  9900|          [36844066]|         1|\n",
      "| 10362|          [65278216]|         1|\n",
      "| 11458|          [39169321]|         1|\n",
      "+------+--------------------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "reversedGraph.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- friend: integer (nullable = true)\n",
      " |-- users: array (nullable = true)\n",
      " |    |-- element: integer (containsNull = true)\n",
      " |-- users_size: integer (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "reversedGraph.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[summary: string, friend: string, users_size: string]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "reversedGraph.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "window = Window.orderBy(col(\"users_size\").desc())\n",
    "    \n",
    "top50 = reversedGraph.withColumn(\"row_number\", row_number().over(window)) \\\n",
    "            .filter(col(\"row_number\") < 50) \\\n",
    "            .select(col(\"friend\"), col(\"users_size\")) \\\n",
    "            .orderBy(col(\"users_size\").desc()) \\\n",
    "            .collect()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "9606655 244\n",
      "62922315 241\n",
      "1288836 240\n",
      "36402159 239\n",
      "36079654 239\n",
      "40342046 235\n",
      "24319760 234\n",
      "34854364 234\n",
      "45353567 233\n",
      "28229916 231\n",
      "16364918 230\n",
      "52511791 229\n",
      "549319 227\n",
      "5137947 227\n",
      "65079230 227\n",
      "17636074 226\n",
      "49067109 225\n",
      "53106903 225\n",
      "6570168 223\n",
      "44621704 223\n",
      "34850500 223\n",
      "27338193 222\n",
      "32810368 222\n",
      "25606717 222\n",
      "34201873 220\n",
      "6147442 219\n",
      "62386165 219\n",
      "45239367 219\n",
      "32821462 218\n",
      "30234171 218\n",
      "63649194 217\n",
      "53826156 217\n",
      "13813472 217\n",
      "26158314 217\n",
      "17679500 217\n",
      "14394422 216\n",
      "7153815 216\n",
      "13062446 216\n",
      "36039499 216\n",
      "64373911 216\n",
      "12890141 215\n",
      "20291955 215\n",
      "36757249 214\n",
      "64856469 214\n",
      "40043869 213\n",
      "34071175 212\n",
      "11768267 211\n",
      "38750752 211\n",
      "3295906 211\n"
     ]
    }
   ],
   "source": [
    "for val in top50:\n",
    "    print '%s %s' % val"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}