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

本文涉及的产品
云原生大数据计算服务 MaxCompute,5000CU*H 100GB 3个月
云原生大数据计算服务MaxCompute,500CU*H 100GB 3个月
简介: 大数据-106 Spark Graph X 计算学习 案例:1图的基本计算、2连通图算法、3寻找相同的用户

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

目前已经更新到了:

Hadoop(已更完)

HDFS(已更完)

MapReduce(已更完)

Hive(已更完)

Flume(已更完)

Sqoop(已更完)

Zookeeper(已更完)

HBase(已更完)

Redis (已更完)

Kafka(已更完)

Spark(正在更新!)

章节内容

上节完成了如下的内容:


Spark Graph X

基本概述

架构基础

概念详解

核心数据结构

c92a7c92394e104f880ed9bbcfba1876_d13f54fbe15f4376ac954d11486a02ff.png 编写 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的操作

70c42e4d39ecea331248a7e195ee156e_17c1a55c88ce4e32ad53ff6481b6be7f.png

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

案例二:连通图算法

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

数据内容

自己生成一些即可:

1 2
1 3
2 4
3 4
4 5
5 6

编写代码

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))

运行的截图如下图:

相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
目录
相关文章
|
1月前
|
分布式计算 大数据 Apache
ClickHouse与大数据生态集成:Spark & Flink 实战
【10月更文挑战第26天】在当今这个数据爆炸的时代,能够高效地处理和分析海量数据成为了企业和组织提升竞争力的关键。作为一款高性能的列式数据库系统,ClickHouse 在大数据分析领域展现出了卓越的能力。然而,为了充分利用ClickHouse的优势,将其与现有的大数据处理框架(如Apache Spark和Apache Flink)进行集成变得尤为重要。本文将从我个人的角度出发,探讨如何通过这些技术的结合,实现对大规模数据的实时处理和分析。
105 2
ClickHouse与大数据生态集成:Spark & Flink 实战
|
1月前
|
存储 分布式计算 Hadoop
数据湖技术:Hadoop与Spark在大数据处理中的协同作用
【10月更文挑战第27天】在大数据时代,数据湖技术凭借其灵活性和成本效益成为企业存储和分析大规模异构数据的首选。Hadoop和Spark作为数据湖技术的核心组件,通过HDFS存储数据和Spark进行高效计算,实现了数据处理的优化。本文探讨了Hadoop与Spark的最佳实践,包括数据存储、处理、安全和可视化等方面,展示了它们在实际应用中的协同效应。
94 2
|
1月前
|
存储 分布式计算 Hadoop
数据湖技术:Hadoop与Spark在大数据处理中的协同作用
【10月更文挑战第26天】本文详细探讨了Hadoop与Spark在大数据处理中的协同作用,通过具体案例展示了两者的最佳实践。Hadoop的HDFS和MapReduce负责数据存储和预处理,确保高可靠性和容错性;Spark则凭借其高性能和丰富的API,进行深度分析和机器学习,实现高效的批处理和实时处理。
65 1
|
1月前
|
分布式计算 Java 开发工具
阿里云MaxCompute-XGBoost on Spark 极限梯度提升算法的分布式训练与模型持久化oss的实现与代码浅析
本文介绍了XGBoost在MaxCompute+OSS架构下模型持久化遇到的问题及其解决方案。首先简要介绍了XGBoost的特点和应用场景,随后详细描述了客户在将XGBoost on Spark任务从HDFS迁移到OSS时遇到的异常情况。通过分析异常堆栈和源代码,发现使用的`nativeBooster.saveModel`方法不支持OSS路径,而使用`write.overwrite().save`方法则能成功保存模型。最后提供了完整的Scala代码示例、Maven配置和提交命令,帮助用户顺利迁移模型存储路径。
|
2月前
|
消息中间件 分布式计算 NoSQL
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
大数据-104 Spark Streaming Kafka Offset Scala实现Redis管理Offset并更新
43 0
|
2月前
|
消息中间件 存储 分布式计算
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
大数据-103 Spark Streaming Kafka Offset管理详解 Scala自定义Offset
98 0
|
1月前
|
SQL 机器学习/深度学习 分布式计算
Spark快速上手:揭秘大数据处理的高效秘密,让你轻松应对海量数据
【10月更文挑战第25天】本文全面介绍了大数据处理框架 Spark,涵盖其基本概念、安装配置、编程模型及实际应用。Spark 是一个高效的分布式计算平台,支持批处理、实时流处理、SQL 查询和机器学习等任务。通过详细的技术综述和示例代码,帮助读者快速掌握 Spark 的核心技能。
58 6
|
1月前
|
分布式计算 大数据 OLAP
AnalyticDB与大数据生态集成:Spark & Flink
【10月更文挑战第25天】在大数据时代,实时数据处理和分析变得越来越重要。AnalyticDB(ADB)是阿里云推出的一款完全托管的实时数据仓库服务,支持PB级数据的实时分析。为了充分发挥AnalyticDB的潜力,将其与大数据处理工具如Apache Spark和Apache Flink集成是非常必要的。本文将从我个人的角度出发,分享如何将AnalyticDB与Spark和Flink集成,构建端到端的大数据处理流水线,实现数据的实时分析和处理。
62 1
|
2月前
|
分布式计算 大数据 Apache
利用.NET进行大数据处理:Apache Spark与.NET for Apache Spark
【10月更文挑战第15天】随着大数据成为企业决策和技术创新的关键驱动力,Apache Spark作为高效的大数据处理引擎,广受青睐。然而,.NET开发者面临使用Spark的门槛。本文介绍.NET for Apache Spark,展示如何通过C#和F#等.NET语言,结合Spark的强大功能进行大数据处理,简化开发流程并提升效率。示例代码演示了读取CSV文件及统计分析的基本操作,突显了.NET for Apache Spark的易用性和强大功能。
50 1
|
2月前
|
消息中间件 分布式计算 Kafka
大数据平台的毕业设计02:Spark与实时计算
大数据平台的毕业设计02:Spark与实时计算
103 0