GCP-BigQuery-PageRank.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Package Install\n",
    "```\n",
    "$ sudo apt-get install python2.7-dev\n",
    "$ sudo easy_install pip\n",
    "$ pip install jupyter\n",
    "$ pip install pandas\n",
    "$ pip install google-api-python-client\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Jupyter spark enviroment.\n",
    "```\n",
    "export PYSPARK_DRIVER_PYTHON=jupyter\n",
    "export PYSPARK_DRIVER_PYTHON_OPTS=\"notebook\"\n",
    "pyspark --master local[2] --packages graphframes:graphframes:0.3.0-spark2.0-s_2.11\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requesting query... ok.\n",
      "Query running...\n",
      "Query done.\n",
      "Cache hit.\n",
      "\n",
      "Retrieving results...\n",
      "Got 100000 rows.\n",
      "\n",
      "Total time taken 6.56 s.\n",
      "Finished at 2017-01-12 09:16:12.\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "\n",
    "projectId = 'spark-151209'\n",
    "query = \"\"\"SELECT Actor1CountryCode, Actor2CountryCode, EventRootCode FROM [gdelt-bq:full.events] \n",
    "WHERE Actor1CountryCode != \"\"\n",
    "AND Actor2CountryCode != \"\"\n",
    "AND EventRootCode = '19'\n",
    "LIMIT 100000\"\"\"\n",
    "df = pd.read_gbq(query, projectId)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------+-----------------+-------------+\n",
      "|Actor1CountryCode|Actor2CountryCode|EventRootCode|\n",
      "+-----------------+-----------------+-------------+\n",
      "|              USA|              USA|           19|\n",
      "|              USA|              NZL|           19|\n",
      "|              USA|              USA|           19|\n",
      "|              CAN|              POL|           19|\n",
      "|              USA|              USA|           19|\n",
      "|              USA|              USA|           19|\n",
      "|              ZAF|              ZAF|           19|\n",
      "|              GBR|              USA|           19|\n",
      "|              USA|              USA|           19|\n",
      "|              USA|              USA|           19|\n",
      "|              USA|              USA|           19|\n",
      "|              BGD|              BGD|           19|\n",
      "|              USA|              USA|           19|\n",
      "|              NZL|              NZL|           19|\n",
      "|              UKR|              POL|           19|\n",
      "|              SAU|              ARE|           19|\n",
      "|              SAU|              ARE|           19|\n",
      "|              AFG|              IRN|           19|\n",
      "|              AFG|              AFG|           19|\n",
      "|              ITA|              USA|           19|\n",
      "+-----------------+-----------------+-------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "sqlctx = SQLContext(sc)\n",
    "sdf = sqlctx.createDataFrame(df)\n",
    "sdf.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import graphframes\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "keys = sdf.rdd.flatMap(lambda x: (x[0], x[1])).distinct()\n",
    "keylist = keys.collect()\n",
    "vertices = keys.map(lambda x: (x,)).toDF([\"id\"])\n",
    "edge = sdf.rdd.map(lambda x: (x[0], x[1], x[0] + \":\" +x[1])).toDF([\"src\", \"dst\", \"relation\"])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "g = graphframes.GraphFrame(vertices, edge)\n",
    "results = g.pageRank(resetProbability=0.15, tol=0.01)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---+------------------+\n",
      "| id|          pagerank|\n",
      "+---+------------------+\n",
      "|USA|31.021184071302333|\n",
      "|SYR| 8.286111156778835|\n",
      "|GBR| 6.005561114899255|\n",
      "|AFG| 5.162184088030115|\n",
      "|RUS| 3.934112277405712|\n",
      "|IRQ| 3.879846182250965|\n",
      "|ISR|3.8232246351402797|\n",
      "|FRA|3.5474286759348614|\n",
      "|TUR|3.3192346271949136|\n",
      "|DEU|3.1362004218725374|\n",
      "|PAK| 3.021622028295468|\n",
      "|AFR|2.7670425768775684|\n",
      "|AUS| 2.740062004184414|\n",
      "|CHN|2.7033085585495398|\n",
      "|IRN| 2.332781842266423|\n",
      "|NGA| 2.202476105742932|\n",
      "|JPN| 2.175917593090732|\n",
      "|SAU|  2.11733321954107|\n",
      "|PSE|2.1140862538489116|\n",
      "|CAN|1.9687376455852512|\n",
      "+---+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "NoneType"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "text = results.vertices.sort('pagerank', ascending=False).select(\"id\", \"pagerank\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "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.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}