course2-big-data-analysis/c02w06_BFS_Spark_SQL.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Breadth-first search in Spark SQL\n",
    "\n",
    "In the first course of the specialization you've already implemented BFS (Breadth-first search) using the RDD API. In this assignment you will implement the same algorithm using the Dataframe API.\n",
    "\n",
    "The point of this assignment is to see in practice how fast Spark SQL is and why is this the default API in Spark now.\n",
    "\n",
    "Your goal is to compute the length of the shortest path between two vertices. But now your implementation will be tested against the dataset of the greater size. Notice, that the answer will change because the graph is more dense now.\n",
    "\n",
    "It is instructive to remember the implementation of the algorithm in Spark Core:\n",
    "\n",
    "    def parse_edge(s):\n",
    "      user, follower = s.split(\"\\t\")\n",
    "      return (int(user), int(follower))\n",
    "\n",
    "    def step(item):\n",
    "      prev_v, prev_d, next_v = item[0], item[1][0], item[1][1]\n",
    "      return (next_v, prev_d + 1)\n",
    "\n",
    "    def complete(item):\n",
    "      v, old_d, new_d = item[0], item[1][0], item[1][1]\n",
    "      return (v, old_d if old_d is not None else new_d)\n",
    "\n",
    "    n = 400  # number of partitions\n",
    "    edges = sc.textFile(\"/data/twitter/twitter_sample.txt\").map(parse_edge).cache()\n",
    "    forward_edges = edges.map(lambda e: (e[1], e[0])).partitionBy(n).persist()\n",
    "\n",
    "    x = 12\n",
    "    d = 0\n",
    "    distances = sc.parallelize([(x, d)]).partitionBy(n)\n",
    "    while True:\n",
    "      candidates = distances.join(forward_edges, n).map(step)\n",
    "      new_distances = distances.fullOuterJoin(candidates, n).map(complete, True).persist()\n",
    "      count = new_distances.filter(lambda i: i[1] == d + 1).count()\n",
    "      if count > 0:\n",
    "        d += 1\n",
    "        distances = new_distances\n",
    "        print(\"d = \", d, \"count = \", count)\n",
    "      else:\n",
    "        break\n",
    "\n",
    "Your goal is to implement the same algorithm, using Spark SQL. Keep in mind that you should avoid using UDFs, if you are stuck, take a look at pyspark.sql.functions module. You will definitely need it.\n",
    "\n",
    "Your task is to find the shortest path between vertices 12 and 34. In case of multiple shortest paths, the first one will suffice. Output format is a single number which is the length of the shortest path.\n",
    "\n",
    "    12\n",
    "\n",
    "The result on the sample dataset:\n",
    "    \n",
    "    8\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField, IntegerType\n",
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder.enableHiveSupport().master(\"local[2]\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "graph_schema = StructType([\n",
    "    StructField(\"to_v\", IntegerType(), False),\n",
    "    StructField(\"from_v\", IntegerType(), False)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "dist_schema = StructType([\n",
    "    StructField(\"vertex\", IntegerType(), False),\n",
    "    StructField(\"distance\", IntegerType(), False)\n",
    "])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def shortest_path(v_from, v_to, dataset_path=None):\n",
    "\n",
    "    edges = spark.read.csv(dataset_path, sep=\"\\t\", schema=graph_schema)       \n",
    "    edges.cache()\n",
    "\n",
    "    distances = spark.createDataFrame([(v_from, 0)], dist_schema)\n",
    "    d = 0\n",
    "    while True:\n",
    "        candidates = (distances\n",
    "                      .join(edges, distances.vertex==edges.from_v)\n",
    "                      .select(edges.to_v.alias(\"vertex\"), (distances.distance+1).alias(\"distance\")) \n",
    "                     ).cache()\n",
    "\n",
    "        new_distances = (distances\n",
    "                         .join(candidates, on=\"vertex\", how=\"full_outer\")\n",
    "                         .select(\"vertex\",\n",
    "                                 when(\n",
    "                                     distances.distance.isNotNull(), distances.distance\n",
    "                                 ).otherwise(\n",
    "                                     candidates.distance\n",
    "                                 ).alias(\"distance\"))\n",
    "                        ).persist()\n",
    "        \n",
    "        count = new_distances.where(new_distances.distance==d+1).count()\n",
    "        \n",
    "        if count > 0:\n",
    "            d += 1            \n",
    "            distances = candidates\n",
    "        else:\n",
    "            break  \n",
    "            \n",
    "        target = (new_distances\n",
    "                  .where(new_distances.vertex == v_to)\n",
    "                 ).count()\n",
    "        \n",
    "        if  target > 0:\n",
    "            break\n",
    "\n",
    "    return d"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "8\n"
     ]
    }
   ],
   "source": [
    "d = shortest_path(12, 34, \"/data/twitter/twitter_sample2.txt\")\n",
    "print(d)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}