基于spark Graph X的图形数据分析

    Graph=（V，E）
    集合V={v1,v2,v3}  //顶点
    集合E={(v1,v2),(v1,v3),(v2,v3)}    //边
    //有向图---所有点到点的距离
    D={V,E}
    V={A,B,C,D,E}
    E={<A,B>,<B,C>,<B,D>,<D,A>,<D,E>}
    //无向图
    G={V,E}
    V={A,B,C,D,E}
    E={(A,B),(A,D),(B,C),(B,D),(C,E),(D,E)}
    //有环图
    包含一些列顶点连接的回路（环路）
    //无环图
    DAG即为有向无环图
    
    //解释专业术语
    度：一个顶点所有边的数量
    出度：指从当前顶点指向其他顶点的边的数量（这个点有多少进）
    入度：其他顶点指向当前顶点的边的数量（这个点有多少出）

    //演示：创建Graph
    
    //先创建一个类
    class Graph[VD,ED]{
    //顶点集合算子
    val vertices:VertexRDD[VD]
    //边集合算子
    val edges:EdgeRDD[ED]
    //图形集合算子
    val triplets:RDD[EdgeTriplet[VD,ED]]
    }
    //先要导包
    import org.apache.spark.graphx._
    val vertices:RDD[(Vertexld,Int) = sc.makeRDD(Seq((1L,1),(2L,2),(3L,3)))]
    //1L,2L表示点（可以是对象），1，2表示权重（可以是对象），Edge表示边
    val edges = sc.makeRDD(Seq(Edge(1L,2L,1),Edge(2L,3L,2)))
    //构建关系网使用Grapg(_,_)
    val graph=Graph(vertices,edges)  //Graph[Int,Int] ?
    //读文件
    import org.zpzche.spark.graphx.GraphLoader
    //加载边列表文件创建图，文件每行描述一条边，格式srcld dstld。顶点与边的属性均为1
    val graph = GraphLoader.edgerListFile(sc,"file:///....")
    
    //查看图信息
    顶点数量，边数量，度，入度，出度
    class Graph[VD,ED]{
    val numEdges:Long
    val numVertices:Long
    val inDegrees:VertexRDD[Int]
    val degrees:VertexRDD[Int]
    }

    //使用subList(_,_)从大集合里面截取小集合
    import java.util.ArrayList;
    
    public class MyTest1{
    ArrayList lst = new ArrayList();
    for(int i+1;i<=100;i++){
        lst.add(i)
    }
    System.out.ptitln(lst.subList(50,100));
    }
    结果：   本次特别指出：subList(_,_)的作用
    [51,52,53,....100]
    
    引出：subgraph(_,_)的作用：从大图里面截取小子图，图是非线性。
    class Graph[VD, ED] {
      //构建图
      //改变边的方向
      def reverse: Graph[VD, ED]
      //从大图里截取子图，然后根据条件返回为true留下/false抛弃，生成满足顶点与边的条件子图
      def subgraph(epred: EdgeTriplet[VD,ED] => Boolean,
               vpred: (VertexId, VD) => Boolean): Graph[VD, ED]
      }
      
    val t1_graph = tweeter_graph.reverse
    val t2_graph = tweeter_graph.subgraph(vpred=(id,attr)=>attr._2<65)  //attr:(name,age)

    //需求说明
    /*
    数据：twitter-graph-data.txt
    格式：((User*, *),(User*,*))
    (User*, *)=(用户名,用户ID)
    第一个用户表示被跟随者（followee）
    第二个用户表示跟随者（follower）
    创建图并计算每个用户的粉丝数量
    找出网络红人
    */
    //需要数据清洗（数据抽取）时间消耗最长，一个项目数据清洗需要三五个月乃至更长时间。
    //数据分析（模型计算）需要时间也很长。
    思路解说：方向：拿出所有数据，先变成点集合，然后在变成边集合
    ((User47,86566510),(User83,15647839))
    ((User47,86566510),(User42,197134784))
    ((User89,74286565),(User49,19315174))
    ((User16,22679419),(User69,45705189))
    ......
    读出文件，拿数据切割，去重，按照逗号分割，把括号去掉，使用元组
    //导包
    scala> import org.apache.spark.graphx._ 
    scala> val begin=sc.textFile("file:///opt/data/twitter_graph_data.txt")
    //计算顶点
    val vects = begin.map(e=>{
    var arr = e.replace("(","").replace(")","").split(",");
    Seq((arr(1).toLong,arr(0)),(arr(3).toLong,arr(2)))
    }).flatMap(e=>e).distinct(6)
    
    val lines = begin.map(e=>{
    var arr = e.replace("(","").replace(")","").split(",");
    Edge(arr(1).toLong,arr(3).toLong,1)
    })
    
    scala> val graph = Graph(vects,lines)
    //查看所有入度
    scala> graph.inDegrees.count
    res2: Long = 97                                                                 
    
    //遍历所有点
    scala> graph.inDegrees.((id,attr)=>)

    //小细节：
    查看所有命令：  wc -l _.txt
    查看文件首行：  head -n 2 _.txt  //显示两行
    查看文件尾行：  tail -n 2 _.txt  //显示两行
    显示文件行号：  set nu
    启动几个分区：  .distinct(_)
    数据采集：	   .collect
    数据计算：	   .count
    二维数组压平：   .flatMap(_=>_)
    限制：			.limit()
    cat主要有三大功能：
    1.一次显示整个文件。
    $ cat   filename
    2.从键盘创建一个文件。
    $ cat  >  filename
    只能创建新文件,不能编辑已有文件.
    3.将几个文件合并为一个文件。
    $cat   file1   file2  > file
    //显示详细的表中前五条信息
    df.show(5,flase)
    //文件传输本地(虚拟机)
    load("file:///root/a.csv")
    //文件传输hdfs中
    load("hdfs:///root/a.csv") = load(/root/a.csv)
    //远程hdfs连接
    load("hdfs://sandbox-hdp.hortonworks.com:8020/root/a.csv")

    //官方解释map()与flatMap()的区别
    
    //map:Returns a stream consisting of the results of applying the given function to the elements of this stream.
    
    //返回一个流，包含给定函数应用在流中每一个元素后的结果
    
    //flatmap:Returns a stream consisting of the results of replacing each element of this stream with the contents of a mapped stream produced by applying the provided mapping function to each element.
    
    //返回一个流，包含将此流中的每个元素替换为通过给定函数映射应用于每个元素而生成的映射流的内容
    
    //举例说明
    
    //有二箱鸡蛋，每箱5个，现在要把鸡蛋加工成煎蛋，然后分给学生。
    
    //map做的事情：把二箱鸡蛋分别加工成煎蛋，还是放成原来的两箱，分给2组学生；
    
    //flatMap做的事情：把二箱鸡蛋分别加工成煎蛋，然后放到一起【10个煎蛋】，分给10个学生；（压扁，去重功能）