Nebula AlgorithmGraph
Graph的Spark应用程序,通过提交Spark任务的形式使用完整的算法工具对NebulaGraph数据库中的数据执行图计算,也可以通过编程形式调用lib库下的算法针对DataFrame执行图计算。
前提条件Graph
在使用 Algorithm 之前,用户需要确认以下信息:
- NebulaGraph 服务已经部署并启动。详细信息,参考Graph。
- Spark 版本为 2.4.x。
- Scala 版本为 2.11。
- (可选)如果用户需要在Github中克隆最新的Algorithm,并自行编译打包,可以选择安装Graph。
使用限制Graph
点ID的数据必须为整数,即点ID可以是INT类型,或者是String类型但数据本身为整数。
对于非整数的String类型数据,推荐使用调用算法接口的方式,可以使用SparkSQL的dense_rank
函数进行编码,将String类型转换为Long类型。
支持算法Graph
Nebula Algorithm支持的图计算算法如下。
算法名 | 说明 | 应用场景 |
---|---|---|
PageRank | 页面排序 | 网页排序、重点节点挖掘 |
Louvain | 社区发现 | 社团挖掘、层次化聚类 |
KCore | K核 | 社区发现、金融风控 |
LabelPropagation | 标签传播 | 资讯传播、广告推荐、社区发现 |
ConnectedComponent | 联通分量 | 社区发现、孤岛发现 |
StronglyConnectedComponent | 强联通分量 | 社区发现 |
ShortestPath | 最短路径 | 路径规划、网络规划 |
TriangleCount | 三角形计数 | 网络结构分析 |
GraphTriangleCount | 全图三角形计数 | 网络结构及紧密程度分析 |
BetweennessCentrality | 介数中心性 | 关键节点挖掘,节点影响力计算 |
DegreeStatic | 度统计 | 图结构分析 |
实现方法Graph
Nebula Algorithm实现图计算的流程如下:
-
利用Nebula Spark Connector从NebulaGraph数据库中读取图数据为DataFrame。
-
将DataFrame转换为GraphX的图。
-
调用GraphX提供的图算法(例如PageRank)或者自行实现的算法(例如Louvain社区发现)。
详细的实现方法可以参见相关Graph。
获取Nebula AlgorithmGraph
编译打包Graph
-
克隆仓库
nebula-algorithm
。$ git clone -b v2.5 https://github.com/vesoft-inc/nebula-algorithm.git
-
进入目录
nebula-algorithm
。$ cd nebula-algorithm
-
编译打包。
$ mvn clean package -Dgpg.skip -Dmaven.javadoc.skip=true -Dmaven.test.skip=true
编译完成后,在目录nebula-algorithm/target
下生成类似文件nebula-algorithm-2.5.1.jar
。
Maven远程仓库下载Graph
Graph
使用方法Graph
调用算法接口(推荐)Graph
lib
库中提供了10种常用图计算算法,用户可以通过编程调用的形式调用算法。
-
在文件
pom.xml
中添加依赖。<dependency> <groupId>com.vesoft</groupId> <artifactId>nebula-algorithm</artifactId> <version>2.5.1</version> </dependency>
-
传入参数调用算法(以PageRank为例)。更多算法请参见Graph。
Note
执行算法的DataFrame默认第一列是起始点,第二列是目的点,第三列是边权重(非NebulaGraph中的Rank)。
val prConfig = new PRConfig(5, 1.0) val louvainResult = PageRankAlgo.apply(spark, data, prConfig, false)
直接提交算法包Graph
Note
使用封装好的算法包有一定的局限性,例如落库到NebulaGraph时,落库的图空间中创建的Tag的属性名称必须和代码内预设的名称保持一致。如果用户有开发能力,推荐使用第一种方法。
-
设置Graph。
{ # Spark相关配置 spark: { app: { name: LPA # Spark分片数量 partitionNum:100 } master:local } data: { # 数据源,可选值为nebula、csv、json。 source: nebula # 数据落库,即图计算的结果写入的目标,可选值为nebula、csv、json。 sink: nebula # 算法是否需要权重。 hasWeight: false } # NebulaGraph相关配置 nebula: { # 数据源。NebulaGraph作为图计算的数据源时,nebula.read的配置才生效。 read: { # 所有Meta服务的IP地址和端口,多个地址用英文逗号(,)分隔。格式: "ip1:port1,ip2:port2"。 # 使用docker-compose部署,端口需要填写docker-compose映射到外部的端口 # 可以用`docker-compose ps`查看 metaAddress: "192.168.*.10:9559" # NebulaGraph图空间名称 space: basketballplayer # NebulaGraph Edge type, 多个labels时,多个边的数据将合并。 labels: ["serve"] # NebulaGraph每个Edge type的属性名称,此属性将作为算法的权重列,请确保和Edge type对应。 weightCols: ["start_year"] } # 数据落库。图计算结果落库到NebulaGraph时,nebula.write的配置才生效。 write:{ # Graph服务的IP地址和端口, 多个地址用英文逗号(,)分隔。格式: "ip1:port1,ip2:port2"。 # 使用docker-compose部署,端口需要填写docker-compose映射到外部的端口 # 可以用`docker-compose ps`查看 graphAddress: "192.168.*.11:9669" # 所有Meta服务的IP地址和端口,多个地址用英文逗号(,)分隔。格式: "ip1:port1,ip2:port2"。 # 使用docker-compose部署,端口需要填写docker-compose映射到外部的端口 # 可以用`docker-compose ps`查看 metaAddress: "192.168.*.12:9559" user:root pswd:nebula # 在提交图计算任务之前需要自行创建图空间及Tag # NebulaGraph图空间名称 space:nb # NebulaGraph Tag名称,图计算结果会写入该Tag。Tag中的属性名称固定如下: # PageRank:pagerank # Louvain:louvain # ConnectedComponent:cc # StronglyConnectedComponent:scc # LabelPropagation:lpa # ShortestPath:shortestpath # DegreeStatic:degree、inDegree、outDegree # KCore:kcore # TriangleCount:tranglecpunt # BetweennessCentrality:betweennedss tag:pagerank } } local: { # 数据源。图计算的数据源为csv文件或json文件时,local.read的配置才生效。 read:{ filePath: "hdfs://127.0.0.1:9000/edge/work_for.csv" # 如果CSV文件没有表头,使用[_c0, _c1, _c2, ..., _cn]表示其表头,有表头或者是json文件时,直接使用表头名称即可。 # 起始点ID列的表头。 srcId:"_c0" # 目的点ID列的表头。 dstId:"_c1" # 权重列的表头 weight: "_c2" # csv文件是否有表头 header: false # csv文件的分隔符 delimiter:"," } # 数据落库。图计算结果落库到csv文件或text文件时,local.write的配置才生效。 write:{ resultPath:/tmp/ } } algorithm: { # 需要执行的算法,可选值为:pagerank、louvain、connectedcomponent、 # labelpropagation、shortestpaths、degreestatic、kcore、 # stronglyconnectedcomponent、trianglecount、betweenness executeAlgo: pagerank # PageRank参数 pagerank: { maxIter: 10 resetProb: 0.15 # 默认为0.15 } # Louvain参数 louvain: { maxIter: 20 internalIter: 10 tol: 0.5 } # ConnectedComponent/StronglyConnectedComponent参数 connectedcomponent: { maxIter: 20 } # LabelPropagation参数 labelpropagation: { maxIter: 20 } # ShortestPath参数 shortestpaths: { # several vertices to compute the shortest path to all vertices. landmarks: "1" } # DegreeStatic参数 degreestatic: {} # KCore参数 kcore:{ maxIter:10 degree:1 } # TriangleCount参数 trianglecount:{} # BetweennessCentrality参数 betweenness:{ maxIter:5 } } }
-
提交图计算任务。
${SPARK_HOME}/bin/spark-submit --master <mode> --class com.vesoft.nebula.algorithm.Main <nebula-algorithm-2.5.1.jar_path> -p <application.conf_path>
示例:
${SPARK_HOME}/bin/spark-submit --master "local" --class com.vesoft.nebula.algorithm.Main /root/nebula-algorithm/target/nebula-algorithm-2.5.1.jar -p /root/nebula-algorithm/src/main/resources/application.conf
视频Graph
- Graph(2分36秒)