07_pregel求连通组件.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "连通组件（不在乎方向）求解，可以拆解为如下两步：   \n",
    "\n",
    "1.将当前节点的id，发送给周围邻居；   \n",
    "2.对所有聚合后的id集合以及当前id取最小，更新为当前节点属性；  \n",
    "\n",
    "重复1～2直到收敛，案例流程如下：   \n",
    "![avatar](./pic/连通组件.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "sparkConf = org.apache.spark.SparkConf@62d99e07\n",
       "sc = org.apache.spark.SparkContext@d87cb67\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "org.apache.spark.SparkContext@d87cb67"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import org.apache.spark.{SparkConf, SparkContext}\n",
    "import org.apache.spark.graphx._\n",
    "val sparkConf: SparkConf = new SparkConf().setMaster(\"local[2]\")\n",
    "val sc: SparkContext = new SparkContext(sparkConf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "initGraph = org.apache.spark.graphx.impl.GraphImpl@56195ad\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "org.apache.spark.graphx.impl.GraphImpl@56195ad"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val initGraph=util.GraphGenerators.rmatGraph(sc,32,64).removeSelfEdges().mapVertices((vid,_)=>vid)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "rstGraph = org.apache.spark.graphx.impl.GraphImpl@235df905\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "org.apache.spark.graphx.impl.GraphImpl@235df905"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val rstGraph=Pregel(initGraph,initialMsg=Long.MaxValue,activeDirection=EdgeDirection.Either)(\n",
    "    vprog=(_,vd,aggMsg)=>math.min(vd,aggMsg),\n",
    "    sendMsg=(edge)=>{\n",
    "        if(edge.srcAttr<edge.dstAttr){\n",
    "            Iterator((edge.dstId,edge.srcAttr)) \n",
    "        }else if(edge.dstAttr<edge.srcAttr){\n",
    "            Iterator((edge.srcId,edge.dstAttr)) \n",
    "        }else{\n",
    "            Iterator.empty\n",
    "        }\n",
    "    },\n",
    "    mergeMsg=(a,b)=>math.min(a,b)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array((16,16), (22,16), (28,16), (30,16), (24,16), (18,16), (20,16), (26,16), (19,16), (21,16), (25,16), (29,16), (27,16), (23,16), (17,16), (31,16))"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//结果\n",
    "rstGraph.vertices.take(32)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array((16,16), (22,16), (28,16), (30,16), (24,16), (18,16), (20,16), (26,16), (19,16), (21,16), (25,16), (29,16), (27,16), (23,16), (17,16), (31,16))"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//官方api\n",
    "initGraph.connectedComponents().vertices.take(32)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Spark2.4.8 - Scala",
   "language": "scala",
   "name": "spark2.4.8_scala"
  },
  "language_info": {
   "codemirror_mode": "text/x-scala",
   "file_extension": ".scala",
   "mimetype": "text/x-scala",
   "name": "scala",
   "pygments_lexer": "scala",
   "version": "2.11.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}