大数据-106 Spark Graph X 计算学习案例：1图的基本计算、2连通图算法、3寻找相同的用户

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（正在更新！）

章节内容

上节完成了如下的内容：

Spark Graph X
基本概述
架构基础
概念详解
核心数据结构

编写 Spark GraphX 程序注意的事情

数据分区与负载均衡

由于 GraphX 运行在分布式环境中，数据分区策略直接影响到计算性能。合理分区可以减少网络传输和计算开销，提高图计算的效率。要注意图数据的分布情况，避免数据倾斜。

处理大规模数据时的内存管理

GraphX 会对顶点和边的数据进行分区和缓存，但在处理大规模图数据时，内存管理尤为重要。需要注意内存使用情况，合理配置 Spark 的内存参数，避免内存溢出或垃圾回收频繁的问题。

迭代计算的收敛条件

许多图算法（如 PageRank）是基于迭代计算的，因此要合理设置收敛条件（例如迭代次数或结果变化阈值）。过多的迭代会浪费计算资源，过少的迭代可能导致结果不准确。

图的变换和属性操作

在对图进行操作时，特别是更新顶点和边的属性时，要确保变换操作不会导致数据不一致或图结构的破坏。使用 mapVertices、mapEdges 等操作时，要谨慎处理每个顶点和边的属性。

错误处理与调试

在编写分布式程序时，错误处理和调试尤为重要。GraphX 的操作涉及复杂的图结构，调试时应充分利用 Spark 的日志和错误信息，使用小规模数据集进行初步验证，逐步扩展到大规模数据。

数据存储与序列化

GraphX 在处理大规模图数据时，可能需要将数据保存到外部存储中（如 HDFS）。要注意选择合适的数据格式和序列化方式，以保证数据读写的高效性和可靠性。

扩展性与性能优化

在开发 GraphX 应用时，考虑到未来可能的扩展需求，程序设计应具有一定的扩展性。同时，针对性能的优化也是关键，要通过测试和调整参数来找到最佳的执行配置。

编写 Spark GraphX 程序

以下是编写 Spark GraphX 程序的主要步骤：

初始化 SparkContext

创建 SparkConf 和 SparkContext，这是 Spark 应用程序的入口。

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

object GraphXExample {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("GraphX Example").setMaster("local[*]")
    val sc = new SparkContext(conf)
  }
}

构建顶点和边 RDD

顶点和边是构建图的基本元素。我们可以通过 RDD 来定义这些元素

// 顶点RDD (VertexId, 属性)
val vertices: RDD[(VertexId, String)] = sc.parallelize(Array(
  (1L, "Alice"), 
  (2L, "Bob"), 
  (3L, "Charlie"), 
  (4L, "David")
))

// 边RDD (源顶点ID, 目标顶点ID, 属性)
val edges: RDD[Edge[Int]] = sc.parallelize(Array(
  Edge(1L, 2L, 1), 
  Edge(2L, 3L, 1), 
  Edge(3L, 4L, 1), 
  Edge(4L, 1L, 1)
))

构建图 (Graph)

使用顶点和边的 RDD 来构建图。

val graph = Graph(vertices, edges)

进行图操作或算法计算

你可以对图进行各种操作或使用图算法库进行计算。下面的示例是计算 PageRank。

val ranks = graph.pageRank(0.01).vertices

收集和处理结果

通过 collect 或 saveAsTextFile 等方法获取和处理计算结果。

ranks.collect().foreach { case (id, rank) => 
  println(s"Vertex $id has rank: $rank") 
}

关闭 SparkContext

在程序结束时，关闭 SparkContext 以释放资源。

sc.stop()

导入依赖

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-graphx_2.12</artifactId>
  <version>${spark.version}</version>
</dependency>

案例一：图的基本计算

在这里插入图片描述

编写代码

package icu.wzk

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object GraphExample1 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("GraphExample1")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 初始化数据
    // 定义定点（Long，info）
    val vertexArray: Array[(VertexId, (String, Int))] = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50))
    )

    // 定义边（Long，Long，attr）
    val edgeArray: Array[Edge[Int]] = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3),
    )

    // 构造vertexRDD和edgeRDD
    val vertexRDD: RDD[(Long, (String, Int))] = sc.makeRDD(vertexArray)
    val edgeRDD: RDD[Edge[Int]] = sc.makeRDD(edgeArray)

    // 构造图Graph[VD,ED]
    val graph: Graph[(String, Int), Int] = Graph(vertexRDD, edgeRDD)

    // 属性操作实例
    // 找出图中年龄大于30的顶点
    graph.vertices
      .filter {
        case (_, (_, age)) => age > 30
      }
      .foreach(println)

    // 找出图中属性大于5的边
    graph.edges
      .filter {
        edge => edge.attr > 5
      }
      .foreach(println)

    // 列出边属性 > 5 的triplets
    graph.triplets
      .filter(t => t.attr > 5)
      .foreach(println)

    // degrees操作
    // 找出图中最大的出度、入度、度数
    println("==========outDegrees=============")
    graph.outDegrees.foreach(println)
    val outDegrees: (VertexId, Int) = graph.outDegrees
      .reduce {
        (x, y) => if (x._2 > y._2) x else y
      }
    println(s"Out degree: ${outDegrees}")

    println("==========inDegrees=============")
    graph.inDegrees.foreach(println)
    val inDegrees: (VertexId, Int) = graph.inDegrees
      .reduce {
        (x, y) => if (x._2 > y._2) x else y
      }
    println(s"In degree: ${inDegrees}")

    // 转换操作
    // 顶点的转换操作 所有人年龄+10岁
    graph.mapVertices {
      case (id, (name, age)) => (id, (name, age + 10))
    }
      .vertices
      .foreach(println)

    // 边的转换操作 边的属性 * 2
    graph.mapEdges(e => e.attr * 2)
      .edges
      .foreach(println)

    // 结构操作
    // 顶点年龄 > 30的子图
    val subGraph: Graph[(String, Int), Int] = graph.subgraph(vpred = (id, vd) => vd._2 >= 30)
    println("==========SubGraph=============")
    subGraph.vertices.foreach(println)
    subGraph.edges.foreach(println)

    // 连接操作
    println("============连接操作==============")
    // 创建一个新图 顶点VD的数据类型 User，并从Graph做类型转换
    val initialUserGraph: Graph[User, Int] = graph.mapVertices {
      case (_, (name, age)) => User(name, age, 0, 0)
    }
    // initialUserGraph 与 inDegree outDegree 进行 JOIN 修改 inDeg outDeg
    var userGraph: Graph[User, Int] = initialUserGraph
      .outerJoinVertices(initialUserGraph.inDegrees) {
        case (id, u, inDegOut) => User(u.name, u.age, inDegOut.getOrElse(0), u.outDeg)
      }
      .outerJoinVertices(initialUserGraph.outDegrees) {
        case (id, u, outDegOut) => User(u.name, u.age, u.inDeg, outDegOut.getOrElse(0))
      }

    userGraph.vertices.foreach(println)

    // 找到 出度=入度 的人员
    userGraph.vertices
      .filter {
      case (id, u) => u.inDeg == u.outDeg
    }
      .foreach(println)

    // 聚合操作
    // 找到5到各顶点的最短距离
    // 定义源点
    val sourceId: VertexId = 5L
    val initialGraph: Graph[Double, Int] = graph
      .mapVertices((id, _) => if (id == sourceId) 0.0 else Double.PositiveInfinity)
    val sssp: Graph[Double, Int] = initialGraph.pregel(Double.PositiveInfinity)(
      // 两个消息来的时候，取它们当中路径的最小值
      (id, dist, newDist) => math.min(dist, newDist),
      // Send Message 函数
      // 比较 triplet.srcAttr + triplet.attr 和 triplet.dstAttr
      // 如果小于，则发送消息到目的顶点
      triplet => {
        // 计算权重
        if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
          Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
        } else {
          Iterator.empty
        }
      },
      // mergeMsg
      (a, b) => Math.min(a, b)
    )

    println("找到5到各个顶点的最短距离")
    println(sssp.vertices.collect.mkString("\n"))

    sc.stop()

  }
}

case class User(name: String, age: Int, inDeg: Int, outDeg: Int)

运行结果

(5,(Ed,55))
(6,(Fran,50))
(3,(Charlie,65))
(4,(David,42))
Edge(2,1,7)
Edge(5,3,8)
((5,(Ed,55)),(3,(Charlie,65)),8)
((2,(Bob,27)),(1,(Alice,28)),7)
==========outDegrees=============
(5,3)
(3,2)
(2,2)
(4,1)
Out degree: (5,3)
==========inDegrees=============
(4,1)
(2,2)
(1,2)
(6,2)
(3,1)
In degree: (2,2)
(6,(6,(Fran,60)))
(3,(3,(Charlie,75)))
(2,(2,(Bob,37)))
(1,(1,(Alice,38)))
(5,(5,(Ed,65)))
(4,(4,(David,52)))
Edge(3,6,6)
Edge(2,1,14)
Edge(4,1,2)
Edge(5,6,6)
Edge(5,3,16)
Edge(3,2,8)
Edge(2,4,4)
Edge(5,2,4)
==========SubGraph=============
(6,(Fran,50))
(5,(Ed,55))
(3,(Charlie,65))
(4,(David,42))
Edge(5,3,8)
Edge(3,6,3)
Edge(5,6,3)
============连接操作==============
(3,User(Charlie,65,1,2))
(2,User(Bob,27,2,2))
(1,User(Alice,28,2,0))
(6,User(Fran,50,2,0))
(5,User(Ed,55,0,3))
(4,User(David,42,1,1))
(4,User(David,42,1,1))
(2,User(Bob,27,2,2))
找到5到各个顶点的最短距离
(1,5.0)
(2,2.0)
(3,8.0)
(4,4.0)
(5,0.0)
(6,3.0)

Process finished with exit code 0

运行截图如下：
在这里插入图片描述

Pregel API

图本身是递归数据结构，顶点的属性依赖于它们的邻居的属性，这些邻居的属性又依赖于自己的邻居的属性。所以需要重要的算法都是迭代的重新计算每个顶点的属性，直到满足某个确定的条件。
一系列的图并发抽象被提出来用来表达这些迭代算法。
GraphX公开了一个类似Pregel的操作

在这里插入图片描述

vprog：用户定义的顶点运行程序，它所用每一个顶点，负责接收进来的信息，并计算新的顶点值
sendMsg：发送消息
mergeMsg：合并消息

案例二：连通图算法

给定数据文件，找到存在的连通体

数据内容

自己生成一些即可：

编写代码

package icu.wzk

import org.apache.spark.graphx.{Graph, GraphLoader}
import org.apache.spark.{SparkConf, SparkContext}

object GraphExample2 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("GraphExample2")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 从数据文件中加载 生成图
    val graph: Graph[Int, Int] = GraphLoader.edgeListFile(sc, "graph.txt")
    graph.vertices.foreach(println)
    graph.edges.foreach(println)
    // 生成连通图
    graph.connectedComponents()
      .vertices
      .sortBy(_._2)
      .foreach(println)

    // 关闭 SparkContext
    sc.stop()
  }
}

运行结果

(1,1)
(3,1)
(4,1)
(5,1)
(6,1)
(2,1)
Edge(1,2,1)
Edge(1,3,1)
Edge(2,4,1)
Edge(3,4,1)
Edge(4,5,1)
Edge(5,6,1)
(4,1)
(6,1)
(2,1)
(1,1)
(3,1)
(5,1)

运行截图如下所示：
在这里插入图片描述

案例三：寻找相同的用户，合并信息

需求明确

假设：

假设五个不同信息可以作为用户标识，分别：1X，2X，3X，4X，5X
每次可以选择使用若干为字段作为标识
部分标识可能发生变化，如 12变为13 或 24变为25

根据以上规则，判断以下标识是否代表同一用户：

11-21-32、12-22-33（X）
11-21-32、11-21-52（OK）
21-32、11-21-33（OK）
11-21-32、32-48（OK）

问题：在以下数据中，找到同一个用户，合并相同用户的数据

对于用户标识（id）：合并后去重
对于用户的信息：key相同，合并权重

编写代码

package icu.wzk

import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object GraphExample3 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("GraphExample3")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val dataRDD: RDD[(List[Long], List[(String, Double)])] = sc.makeRDD(
      List(
        (List(11L, 21L, 31L), List("kw$北京" -> 1.0, "kw$上海" -> 1.0, "area$中关村" -> 1.0)),
        (List(21L, 32L, 41L), List("kw$上海" -> 1.0, "kw$天津" -> 1.0, "area$回龙观" -> 1.0)),
        (List(41L), List("kw$天津" -> 1.0, "area$中关村" -> 1.0)),
        (List(12L, 22L, 33L), List("kw$大数据" -> 1.0, "kw$spark" -> 1.0, "area$西二旗" -> 1.0)),
        (List(22L, 34L, 44L), List("kw$spark" -> 1.0, "area$五道口" -> 1.0)),
        (List(33L, 53L), List("kw$hive" -> 1.0, "kw$spark" -> 1.0, "area$西二旗" -> 1.0))
      )
    )

    // 1 将标识信息中的每一个元素抽取出来，作为ID
    // 备注1 这里使用了 flatMap 将元素压平
    // 备注2 这里丢掉了标签信息，因为这个RDD主要用于构造顶点、边
    // 备注3 顶点、边的数据要求Long，这个程序修改后才能用在我们的程序中
    val dotRDD: RDD[(VertexId, VertexId)] = dataRDD.flatMap {
      case (allids, _) => allids.map(id => (id, allids.mkString.hashCode.toLong))
    }

    // 2 定义顶点
    val vertexesRDD: RDD[(VertexId, String)] = dotRDD.map {
      case (id, _) => (id, "")
    }
    // 3 定义边(id: 单个标识信息：ids：全部的标识信息)
    val edgesRDD: RDD[Edge[Int]] = dotRDD.map {
      case (id, ids) => Edge(id, ids, 0)
    }
    // 4 生成图
    val graph = Graph(vertexesRDD, edgesRDD)
    // 5 找到强连通体
    val connectRDD: VertexRDD[VertexId] = graph.connectedComponents().vertices;
    // 6 定义中心点的数据
    val centerVertexRDD: RDD[(VertexId, (List[VertexId], List[(String, Double)]))] = dataRDD.map {
      case (allIds, tags) => (allIds.mkString.hashCode.toLong, (allIds, tags))
    }
    // 7 步骤5、6的数据做join 获取需要合并的数据
    val allInfoRDD = connectRDD.join(centerVertexRDD).map {
      case (_, (id2, (allIds, tags))) => (id2, (allIds, tags))
    }
    // 8 数据聚合（将同一个用户的标识、标签放在一起）
    val mergeInfoRDD: RDD[(VertexId, (List[VertexId], List[(String, Double)]))] = allInfoRDD
      .reduceByKey {
        case ((bufferList, bufferMap), (allIds, tags)) =>
          val newList = bufferList ++ allIds

          // map 合并
          val newMap = bufferMap ++ tags
          (newList, newMap)
      }

    // 9 数据合并（allIds去重，tags合并权重）
    val resultRDD: RDD[(List[VertexId], Map[String, Double])] = mergeInfoRDD.map {
      case (key, (allIds, tags)) =>
        val newIds = allIds.distinct
        val newTags = tags.groupBy(x => x._1).mapValues(lst => lst.map(x => x._2).sum)
        (newIds, newTags)
    }

    resultRDD.foreach(println)

    sc.stop()
  }

}

运行结果

(List(21, 32, 41, 11, 31),Map(area$中关村 -> 2.0, kw$北京 -> 1.0, kw$天津 -> 2.0, kw$上海 -> 2.0, area$回龙观 -> 1.0))
(List(22, 34, 44, 12, 33, 53),Map(kw$大数据 -> 1.0, kw$spark -> 3.0, area$五道口 -> 1.0, area$西二旗 -> 2.0, kw$hive -> 1.0))