course1-big-data-esentials/c01w06_ShortestPathTask1.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Reconstructing the path"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this assignment you will use Spark to compute the shortest path between two vertices. In the video, you have learned how to compute the distances between a source vertex and all other vertices in a graph. Now, your task is to reconstruct the shortest path, that is a sequence of vertices connected by the edges.\n",
    "\n",
    "- Dataset location: /data/twitter/twitter_sample_small.txt\n",
    "- Format: user_id \\t follower_id\n",
    "\n",
    "You can start with the code described in \"Starter...\" (see the next self-reading).\n",
    "\n",
    "Your task is to find the shortest path between vertices 12 and 34. In case of multiple shortest paths (that is, disjoint paths with the same length), any will suffice. Output format is the sequence of vertices, delimited by a comma, without spaces. For example, the path “12 -> 42 -> 34” should be printed as:\n",
    "\n",
    "    12,42,34\n",
    "    \n",
    "Hint: before submitting, check your stopping criteria. In BFS, the search was exhaustive, and in this task your program may terminate earlier, thus saving some precious time.\n",
    "\n",
    "The result on the sample dataset:\n",
    "\n",
    "    12,422,53,52,107,20,23,274,34\n",
    "    \n",
    "If you want to deploy the environment on your own machine, please use [bigdatateam/spark-course1](https://hub.docker.com/r/bigdatateam/spark-course1/) Docker container."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#! /usr/bin/env python\n",
    "\n",
    "from pyspark import SparkConf, SparkContext\n",
    "sc = SparkContext(conf=SparkConf().setAppName(\"MyApp\").setMaster(\"local[2]\"))\n",
    "\n",
    "import re"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "n = 150  # number of partitions\n",
    "start, end = 12, 34\n",
    "\n",
    "def parse_edge(s):\n",
    "    \"\"\"Parse raw data 'user\\tfollwer into a tuple'\"\"\"\n",
    "    user, follower = s.split(\"\\t\")\n",
    "    return (int(user), int(follower))\n",
    "\n",
    "def step(item):\n",
    "    # Add one move along the graph\n",
    "    prev_v, prev_d, next_v = item[0], item[1][0], item[1][1]\n",
    "    return (next_v, prev_d + [next_v])\n",
    "\n",
    "edges = sc.textFile(\"/data/twitter/twitter_sample_small.txt\").map(parse_edge).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# We want to find a path from 'start' to 'end' thus require forward edges\n",
    "forward_edges = edges.map(lambda e: (e[1], e[0])).partitionBy(n).persist()\n",
    "\n",
    "# Create a dataset composed of a tuple (current node, path).\n",
    "# We will fill up the array with possible path\n",
    "paths = sc.parallelize([(start, [start])]).partitionBy(n)\n",
    "\n",
    "def found():\n",
    "    return  paths.filter(lambda x: x[0] == end).count()\n",
    "\n",
    "while not found():\n",
    "    paths = paths.join(forward_edges, n).map(step)\n",
    "    \n",
    "paths.cache()\n",
    "\n",
    "final_paths = (paths\n",
    "               .filter(lambda value: value[0] == end)\n",
    "               .map(lambda value: value[1])\n",
    "              ).cache()\n",
    "\n",
    "result = final_paths.take(1)[0]\n",
    "\n",
    "print(','.join(map(str,result)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}