[人工智能]sparkgraph

什么是sparkgragh ： 它是为了为用户建立关系的视图
SparkGraph图计算
基础篇
1. what？
表示数据关系的数据结构
基本元素： 点Vertex，边Edge
Vertex[(VertexId:Long,VertexAttr:Any)]
Edge[(srcVertexId:Long,dstVertexId:Long,attr:Any)]
组合元素： Triplet（源点SrcVertex+边Edge+目标点LstVertex）
类似于RDD
弹性的
分布式的
容错的
2. why？
能够表达复杂的数据关系，并做复杂的图计算
User(userId:Int,name:String,age:Int)
user_relation(fromUserId:Int,toUserId:Int,relation:String)
3. where?
社交
4. how?
val sc：SparkContext = …
val rddVertex = sc.makeRDD(Seq[(VertexId,T]) //创建点RDD T点属性：单个值用基础类型，多个值用样例类
val rddEdge = sc.makeRDD(Seq[Edge(T)]) //创建边RDD T边属性： 单个值用基础类型，多个值用样例类
val gragh = Graph(rddVertex,rddEdge)
属性
graph.vertices //点集合
graph.edges //边集合
graph.triplets //点边点组合集合
gragh.inDegrees //入度 到某点取得记录条数 （？，dstVertexId)
graph.outDegrees //出度 从某点出发的记录条数 (srcVertexId，？)
转换算子
graph.mapTriplets ED2:Graph[VD,ED2]
//针对点属性逐个变形处理，返回新的属性，最终返回基于变性后的点和原来的边的图对象
graph.mapVertices VD2:Graph[VD2,ED] 变一个点出去
graph.mapEdges ED2:Graph[VD,ED2] 变一个边出去
算法篇

package cn.kgc.sparkGragh

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx._


object SparkGraph01 {

  case class Person(name:String,age:Int)

  def main(args: Array[String]): Unit = {
    val CONF = new SparkConf().setMaster("local[*]").setAppName("spark_graphx_02")
    val SC = new SparkContext(CONF)

    //rddVertex
    val rddVertex: RDD[(Long, Person)] = SC.makeRDD(Array(
      (1L, Person("Henry", 22)),
      (2L, Person("Pola", 20)),
      (3L, Person("Ariel", 18)),
      (4L, Person("Jack", 29)),
      (5L, Person("Tom", 31)),
      (6L, Person("Shel", 16)),
      (7L, Person("Niky", 28))
    ))
    //rddEdge
    val rddEdge: RDD[Edge[String]] = SC.makeRDD(Array(
      Edge(1L, 2L, "family"),
      Edge(1L, 3L, "family"),
      Edge(1L, 4L, "teacher"),
      Edge(1L, 5L, "teacher"),
      Edge(1L, 6L, "teacher"),
      Edge(1L, 7L, "teacher"),
      Edge(4L, 5L, "friend"),
      Edge(4L, 6L, "friend"),
      Edge(6L, 7L, "friend"),
      Edge(1L, 4L, "friend"),
      Edge(1L, 5L, "friend")
    ))

    val graph: Graph[Person, String] = Graph(rddVertex, rddEdge).cache()
    //22岁以上的人 点遍历
    //(VertexId,Person)
    //graph.vertices.filter(_._2.age>22).foreach(println)

    //好友关系  关系遍历
    //(srcVertexId,dstVertexId,attr)
    //graph.edges.filter(_.attr.equals("friend")).foreach(println)

    //我的好友的数量
    var ME = 1L

//    val graphMe: Graph[Person, PartitionID] = graph.mapTriplets(e => {
//      //      e.srcId     //源点VertexId
//      //      e.srcAttr   //源点    --Person
//      //      e.dstId     //目标点   --VertexId
//      //      e.dstAttr   //目标点   --Person
//      //      e.attr      //关系数据   String
//      if (e.srcId == 1L && e.attr.equals("friend")) {
//        1
//      } else {
//        0
//      }
//    })
//    graphMe.triplets.foreach(e=>println(s"${e.srcId}\t${e.srcAttr}\t${e.attr}"))
//    println("____")
//    println(graphMe.triplets.map(_.attr).sum())

//    //aggregateMessages[Int]  Int=>消息的类型
//    graph.aggregateMessages[Int](sendMsg = ec=>{
//      //类似于Triplet,扩展了发送消息的函数
//      //ec.sendToSrc(msg:Int) 向源点发送消息
//      //ec.sendToDst(msg:Int) 向目标点发送消息
//      if(ec.srcId == ME && ec.attr.equals("teacher")){
//        ec.sendToSrc(1)
//      }
//    },mergeMsg = _+_).foreach(println)
//    //(1,2) 1号用户有2个好友

    

    SC.stop()
  }

}