Spark 组件 GraphX、Streaming

Spark 组件 GraphX、Streaming

• 一、Spark GraphX
  
  
  
  • 1.1 GraphX 的主要概念
    
  • 1.2 GraphX 的焦点操纵
    
  • 1.3 示例代码
    
  • 1.4 GraphX 的应用场景
    
  
  
• 二、Spark Streaming
  
  
  
  • 2.1 Spark Streaming 的主要概念
    
  • 2.2 示例代码
    
  • 2.3 Spark Streaming 的集成
    
  • 2.4 Spark Streaming 的应用场景
    
  
  

   Spark GraphX 用于处理图和图并行计算。GraphX 将图的表达和操纵嵌入到 Spark 的数据流 API 中，允许用户在图上实行高效的并行计算。GraphX 结合了图计算和数据流计算的功能，使得它可以或许处理复杂的数据分析任务。
  Spark Streaming 用于处理实时数据流。它允许开发者以微批处理（Micro-batch）的方式处理实时数据，提供了一个高层次的 API，可以轻松地将批处理操纵应用于实时数据流。
  一、Spark GraphX

1.1 GraphX 的主要概念

• 极点 (Vertex)：
  
  
  
  • 图中的节点，表示实体或对象。每个极点都有一个唯一的标识符（ID）和属性。
    
  
  
• 边 (Edge)：
  
  
  
  • 图中的连接，表示极点之间的关系。每条边连接两个极点，而且也可以有属性。
    
  
  
• 图 (Graph)：
  
  
  
  • 由极点和边组成的结构。GraphX 利用 Graph 类来表示图，极点和边的聚集分别由 RDD[VertexId, VD] 和 RDD[Edge[ED]] 表示，此中 VD 和 ED 是极点和边的属性类型。
    
  
  
• Triplet：
  
  
  
  • GraphX 中的 EdgeTriplet 代表一条边及其连接的两个极点的信息，允许同时访问极点和边的属性。
    
  
  

1.2 GraphX 的焦点操纵

• 图构造 (Graph Construction)：
  
  
  
  • 通过极点和边的 RDD 来构建图。比方，利用 Graph(vertices, edges) 构造一个图。
    
  
  
• 图转换 (Graph Transformation)：
  
  
  
  • 对图进行操纵，比方过滤极点和边 (subgraph)，或将极点和边的属性映射到新属性 (mapVertices 和 mapEdges)。
    
  
  
• 聚合消息 (Aggregate Messages)：
  
  
  
  • 用于从邻接极点或边聚合信息。这在实现图算法（如 PageRank）时特殊有效。
    
  
  
• 图算法 (Graph Algorithms)：
  
  
  
  • GraphX 提供了一些预定义的图算法，如 PageRank、Connected Components、Shortest Paths 和 Triangle Counting。
    
  
  

1.3 示例代码

1.3.1 简朴的 GraphX 示例，创建一个图并运行 PageRank 算法：

1. import org.apache.spark.graphx._
2. import org.apache.spark.rdd.RDD
3. import org.apache.spark.sql.SparkSession
5. object GraphXExample {
6.   def main(args: Array[String]): Unit = {
7.     val conf = new SparkConf()
8.       .setAppName("graphx example")
9.       .setMaster("local[*]")
10.    
11.     // 创建 SparkSession
12.     val spark = SparkSession
13.       .builder
14.       .config(conf)
15.       .getOrCreate()
17.     // 创建顶点 RDD
18.     val vertices: RDD[(VertexId, String)] = spark.sparkContext.parallelize(Seq(
19.       (1L, "Alice"),
20.       (2L, "Bob"),
21.       (3L, "Charlie"),
22.       (4L, "David")
23.     ))
25.     // 创建边 RDD
26.     val edges: RDD[Edge[Int]] = spark.sparkContext.parallelize(Seq(
27.       Edge(1L, 2L, 1),
28.       Edge(2L, 3L, 1),
29.       Edge(3L, 4L, 1),
30.       Edge(4L, 1L, 1)
31.     ))
33.     // 创建图
34.     val graph = Graph(vertices, edges)
36.     // 运行 PageRank 算法
37.     // PageRank 算法最初是用于搜索引擎中对网页进行排序的基础算法，通过分析网络中的链接结构来评估每个网页的相对重要性。
38.     // 这里0.0001为收敛阈值
39.     val ranks = graph.pageRank(0.0001).vertices
41.     // 打印结果
42.     ranks.collect().foreach { case (id, rank) => println(s"Vertex $id has rank $rank") }
44.     // 停止 SparkSession
45.     spark.stop()
46.   }
47. }

复制代码

1.3.2 Spark GraphX 的一些基本操纵和概念

1. // 创建图
2. val graph = Graph(vertices, edges)
4. // vertices：获取图中的所有顶点。
5. graph.vertices.collect.foreach(println)
7. // mapVertices：对顶点的属性进行变换。
8. val newGraph = graph.mapVertices((id, attr) => attr.toUpperCase)
10. // edges：获取图中的所有边。
11. graph.edges.collect.foreach(println)
13. // mapEdges：对边的属性进行变换。
14. val newGraph = graph.mapEdges(e => e.attr * 2)
16. // 查看所有的边（将点带入）【完整】
17. graph.triplets.foreach(println)
18. // 入度：指向自己的个数
19. graph.inDegrees.foreach(println) // (人id,入度)
21. // 出度：指向别人的个数
22. graph.outDegrees.foreach(println) // (人id,出度)
24. // 将入度和出度一起显示
25. graph
26.     .inDegrees
27.     .join(graph.outDegrees)
28.     .foreach(println) // (人id,(入度,出度))
30. // 度：入度+出度
31. graph.degrees.foreach(println)
33. // subgraph：根据条件创建一个子图，保留满足条件的顶点和边。
34. val subGraph = graph.subgraph(vpred = (id, attr) => attr != "Bob")
36. // joinVertices：将图的顶点属性与一个新的 RDD 进行连接，并更新顶点属性。
37. val newAttrs: RDD[(Long, String)] = sc.parallelize(Seq(
38.   (1L, "Alice_new"),
39.   (4L, "David_new")
40. ))
42. val joinedGraph = graph.joinVertices(newAttrs) {
43.   case (id, oldAttr, newAttr) => newAttr
44. }

复制代码

1.3.3 图算法

1. // PageRank 算法用于计算图中每个顶点的重要性。
2. val ranks = graph.pageRank(0.0001).vertices
3. ranks.collect.foreach(println)
5. // 连接组件: 连通组件算法用于查找图中的连通子图。
6. val connectedComponents = graph.connectedComponents().vertices
7. connectedComponents.collect.foreach(println)
9. // 三角计数: 三角计数算法用于计算每个顶点所属的三角形数量。
10. val triangleCounts = graph.triangleCount().vertices
11. triangleCounts.collect.foreach(println)
13. // 图的持久化与加载: 图可以通过将顶点和边的 RDD 存储在 HDFS 或其他文件系统中进行持久化。
14. // 保存图的顶点和边
15. graph.vertices.saveAsTextFile("hdfs://path/to/vertices")
16. graph.edges.saveAsTextFile("hdfs://path/to/edges")
18. // 从文件中加载图
19. val loadedVertices: RDD[(Long, String)] = sc.textFile("hdfs://path/to/vertices").map { line =>
20.   val parts = line.split(",")
21.   (parts(0).toLong, parts(1))
22. }
24. val loadedEdges: RDD[Edge[Int]] = sc.textFile("hdfs://path/to/edges").map { line =>
25.   val parts = line.split(",")
26.   Edge(parts(0).toLong, parts(1).toLong, parts(2).toInt)
27. }
29. val loadedGraph: Graph[String, Int] = Graph(loadedVertices, loadedEdges)

复制代码

1.3.4 Pregel 模型

   Pregel 算法是一种迭代、消息通报的计算模型，特殊适用于处理图的遍历和递归题目，如最短路径计算、PageRank、连通组件检测等。
Pregel 的焦点头脑是将图计算任务表示为一个超等步（superstep）序列，每个超等步由以下几个阶段组成：
  

• 消息通报：每个极点可以向相邻的极点发送消息。
  
• 消息处理：每个极点接收来自相邻极点的消息，并更新自己的状态。
  
• 极点计算：极点在处理完消息后可以决定是否继续活跃或制止计算（halt）。
  

  这个过程会不停迭代，直到所有极点都制止计算或达到指定的迭代次数。

1. // Pregel 模型:
2. val initialMsg = ... // 定义初始消息
3. val maxIterations = 10 // 最大迭代次数
5. val resultGraph = graph.pregel(initialMsg, maxIterations)(
6.   // 顶点程序，处理接收到的消息并更新顶点属性
7.   vprog = (id, attr, msg) => {
8.     // 根据顶点的属性和收到的消息，更新顶点属性
9.   },
10.   // 发送消息，定义如何从一个顶点向相邻的顶点发送消息
11.   sendMsg = triplet => {
12.     // 根据边的属性和顶点的状态，决定发送的消息
13.   },
14.   // 聚合消息，定义如何合并一个顶点接收到的所有消息
15.   mergeMsg = (msg1, msg2) => {
16.     // 合并接收到的多个消息
17.   }
18. )

复制代码

利用 Pregel 模型可以计算图中某个出发点到其他所有极点的最短路径。

1. import org.apache.spark.SparkConf
2. import org.apache.spark.graphx._
3. import org.apache.spark.rdd.RDD
4. import org.apache.spark.sql.SparkSession
6. object GraphxTest {
7.   def main(args: Array[String]): Unit = {
9.     val conf = new SparkConf()
10.       .setAppName("graphx example")
11.       .setMaster("local[*]")
13.     // 创建 SparkSession
14.     val spark = SparkSession
15.       .builder
16.       .config(conf)
17.       .getOrCreate()
19.     val vertices: RDD[(VertexId, String)] = spark.sparkContext.parallelize(Seq(
20.       (1L, "Alice"),
21.       (2L, "Bob"),
22.       (3L, "Charlie"),
23.       (4L, "David")
24.     ))
26.     // 创建边 RDD
27.     val edges: RDD[Edge[Int]] = spark.sparkContext.parallelize(Seq(
28.       Edge(1L, 2L, 1),
29.       Edge(2L, 3L, 1),
30.       Edge(2L, 4L, 5),
31.       Edge(3L, 4L, 2),
32.       Edge(4L, 1L, 1)
33.     ))
35.     // 创建图
36.     val graph: Graph[String, PartitionID] = Graph(vertices, edges)
38.     // 设置源顶点 ID
39.     val sourceId: VertexId = 1L
41.     // 初始化图的顶点属性，将源顶点的距离设为 0.0，其他顶点设为无穷大
42.     val initialGraph = graph.mapVertices((id, name) =>
43.       if (id == sourceId) (name, 0.0) else (name, Double.PositiveInfinity)
44.     )
46.     // 运行 Pregel 算法计算单源最短路径 (SSSP)
47.     val sssp = initialGraph.pregel(Double.PositiveInfinity)(
48.       // 顶点程序：更新顶点属性（即距离）为当前距离与新接收到的距离中的较小值
49.       vprog = (id, attr, msg) => (attr._1, Math.min(attr._2, msg)),
51.       // 消息发送函数：计算从源顶点到目标顶点的距离，如果新计算的距离小于目标顶点当前的距离，则发送消息
52.       sendMsg = triplet => {
53.         if (triplet.srcAttr._2 + triplet.attr < triplet.dstAttr._2) {
54.           Iterator((triplet.dstId, triplet.srcAttr._2 + triplet.attr))
55.         } else {
56.           Iterator.empty
57.         }
58.       },
59.       // 消息合并函数：在目标顶点收到多个消息时，取最短的距离
60.       mergeMsg = (a, b) => math.min(a, b)
61.     )
63.     // 打印最终的最短路径结果
64.     sssp.vertices.collect.foreach { case (id, (name, dist)) =>
65.       println(s"Distance from $sourceId to $id ($name) is $dist")
66.     }
68.     // 停止 SparkSession
69.     spark.stop()
70.   }
71. }

复制代码

1.4 GraphX 的应用场景

• 外交网络分析：分析用户之间的关系，如挚友保举、外交影响力分析。
  
• 路径分析：计算最短路径、页面排名等。
  
• 社区检测：辨认图中连接精密的子图或社区。
  
• 网络优化：通过图分析网络拓扑结构，优化数据流路由等。
  

二、Spark Streaming

2.1 Spark Streaming 的主要概念

   Spark Streaming 提供了一个强大且易用的 API，使开发者可以或许轻松地构建实时数据处理应用，特殊适合必要低延长、高吞吐量的场景。
  

• DStream (Discretized Stream)：
  
  
  
  • DStream 是 Spark Streaming 中的焦点抽象，表示一个连续的数据流。它可以被视为一系列 RDD（Resilient Distributed Datasets）的聚集，每个 RDD 都包含某个时间隔断内的数据。
    
  • DStream 可以从各种输入源（如 Kafka、Flume、TCP 套接字等）创建，也可以通过对现有 DStream 的转换来创建。
    
  
  
• Transformations (转换操纵)：
  
  
  
  • 雷同于 Spark 的 RDD 转换操纵，DStream 支持各种转换操纵，如 map、filter、reduceByKey 等。每个转换操纵都会应用于 DStream 中的每个 RDD，生成一个新的 DStream。
    
  
  
• Output Operations (输出操纵)：
  
  
  
  • DStream 提供多种输出操纵，将处理后的数据输出到外部系统。比方，print、saveAsTextFiles、saveAsHadoopFiles、foreachRDD 等。
    
  
  
• Window Operations (窗口操纵)：
  
  
  
  • Spark Streaming 支持窗口操纵，允许对一段时间范围内的数据进行聚合。比方，可以计算已往 10 秒钟内的每 2 秒的数据。
    
  
  

2.2 示例代码

一个简朴的 Spark Streaming 示例，读取 TCP 套接字数据，并进行词频统计：

1. import org.apache.spark.SparkConf
2. import org.apache.spark.streaming.{Seconds, StreamingContext}
4. object SparkStream {
5.   def main(args: Array[String]): Unit = {
6.     // 创建 Spark 配置对象，并设置应用名称和运行模式
7.     val conf = new SparkConf()
8.       .setMaster("local[*]")  // 使用本地模式运行，[*] 表示使用所有可用的 CPU 核心
9.       .setAppName("spark-streaming")  // 设置应用程序名称
11.     // 创建 StreamingContext，设置批处理间隔为 5 秒
12.     val scc = new StreamingContext(conf, Seconds(5))
14.     // 监听本地端口 9999，接收流式数据
15.     scc.socketTextStream("localhost", 9999)
16.       .mapPartitions(
17.         _.flatMap(
18.           _.replaceAll("[^a-zA-Z ]+", "")  
19.             .split("\\s+")
20.             .map((_, 1))
21.         )
22.       )
23.       // 按单词进行归约，计算每个单词的出现次数
24.       .reduceByKey(_ + _)
25.       // 打印结果到控制台
26.       .print()
28.     // 启动 Spark Streaming 计算
29.     scc.start()
30.     // 等待应用程序终止
31.     scc.awaitTermination()
32.   }
33. }

复制代码

2.3 Spark Streaming 的集成

• 与 Kafka 集成：Spark Streaming 可以从 Kafka 中读取数据流，用于实时日记处理、监控等场景。
  
• 与 Flume 集成：结合 Flume 进行分布式日记收集，然后利用 Spark Streaming 实时处理和分析日记。
  
• 与 HDFS、S3 等集成：将处理后的数据输出到 HDFS、S3 等分布式文件系统进行长期化存储。
  
• 与 SQL 和 MLlib 集成：Spark Streaming 可以与 Spark SQL 和 MLlib 集成，进行实时的数据分析和机器学习任务。
  

2.4 Spark Streaming 的应用场景

• 实时日记分析：监控服务器日记、应用日记，检测异常环境。
  
• 实时 ETL (Extract, Transform, Load)：对流式数据进行洗濯、转换，并写入到数据仓库。
  
• 实时监控与报警：对实时数据流进行分析，当检测到特定条件时触发报警。
  
• 在线保举系统：基于实时用户行为数据进行保举，比方在线广告保举。
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。