跳转至

NebulaGraph Algorithm

NebulaGraph Algorithm (简称 Algorithm)是一款基于 GraphX 的 Spark 应用程序,通过提交 Spark 任务的形式使用完整的算法工具对 NebulaGraph 数据库中的数据执行图计算,也可以通过编程形式调用 lib 库下的算法针对 DataFrame 执行图计算。

版本兼容性

NebulaGraph Algorithm 版本和 NebulaGraph 内核的版本对应关系如下。

NebulaGraph 版本 NebulaGraph Algorithm 版本
nightly 3.0-SNAPSHOT
3.0.0 ~ 3.8.x 3.x.0
2.6.x 2.6.x
2.5.0、2.5.1 2.5.0
2.0.0、2.0.1 2.1.0

前提条件

在使用 Algorithm 之前,用户需要确认以下信息:

  • 已安装 Spark2.4 或者 3.x
  • 已安装 Scala 2.11(对应 Spark 2.4)或 2.12(对应 Spark 3.x)。
  • (可选)如果用户需要在 Github 中克隆最新的 Algorithm,并自行编译打包,可以选择安装 Maven

使用限制

图计算会输出点的数据集,算法结果会以DataFrame形式作为点的属性存储。用户可以根据业务需求,自行对算法结果做进一步操作,例如统计、筛选。

Compatibility

Algorithm v3.1.0 版本之前,直接提交算法包时,点 ID 的数据必须为整数,即点 ID 可以是 INT 类型,或者是 String 类型但数据本身为整数。

支持算法

NebulaGraph Algorithm 支持的图计算算法如下。

算法名 说明 应用场景 属性名称 属性数据类型
PageRank 页面排序 网页排序、重点节点挖掘 pagerank double/string
Louvain 鲁汶 社团挖掘、层次化聚类 louvain int/string
KCore K 核 社区发现、金融风控 kcore int/string
LabelPropagation 标签传播 资讯传播、广告推荐、社区发现 lpa int/string
Hanp 标签传播进阶版 社区发现、推荐 hanp int/string
ConnectedComponent 弱联通分量 社区发现、孤岛发现 cc int/string
StronglyConnectedComponent 强联通分量 社区发现 scc int/string
ShortestPath 最短路径 路径规划、网络规划 shortestpath string
TriangleCount 三角形计数 网络结构分析 trianglecount int/string
GraphTriangleCount 全图三角形计数 网络结构及紧密程度分析 count int
BetweennessCentrality 中介中心性 关键节点挖掘,节点影响力计算 betweenness double/string
ClosenessCentrality 紧密中心性 关键节点挖掘、节点影响力计算 closeness double/string
DegreeStatic 度统计 图结构分析 degree,inDegree,outDegree int/string
ClusteringCoefficient 聚集系数 推荐、电信诈骗分析 clustercoefficient double/string
Jaccard 杰卡德相似度计算 相似度计算、推荐 jaccard string
BFS 广度优先遍历 层序遍历、最短路径规划 bfs string
DFS 深度优先遍历 层序遍历、最短路径规划 dfs string
Node2Vec - 图分类 node2vec string

Note

如果需要将算法结果写入到 NebulaGraph 中,请确保对应图空间中的 Tag 有和上表对应的属性名称和数据类型。

实现方法

NebulaGraph Algorithm 实现图计算的流程如下:

  1. 利用 NebulaGraph Spark Connector 从 NebulaGraph 中读取图数据为 DataFrame。

  2. 将 DataFrame 转换为 GraphX 的图。

  3. 调用 GraphX 提供的图算法(例如 PageRank)或者自行实现的算法(例如 Louvain 社区发现)。

详细的实现方法可以参见相关 Scala 文件

获取 NebulaGraph Algorithm

编译打包

  1. 克隆仓库nebula-algorithm

    $ git clone -b v3.2.0 https://github.com/vesoft-inc/nebula-algorithm.git
    

    如果是在 Spark 3.x 环境中运行 Algorithm,请使用 spark3 分支:

    $ git clone -b spark3 https://github.com/vesoft-inc/nebula-algorithm.git
    
  2. 进入目录nebula-algorithm

    $ cd nebula-algorithm
    
  3. 编译打包。

    $ mvn clean package -Dgpg.skip -Dmaven.javadoc.skip=true -Dmaven.test.skip=true
    

编译完成后,在目录nebula-algorithm/target下生成类似文件nebula-algorithm-3.x.x.jar

Maven 远程仓库下载

下载地址

使用方法

Note

如果数据的属性值包含中文字符,可能出现乱码。请在提交 Spark 任务时加上以下选项:

--conf spark.driver.extraJavaOptions=-Dfile.encoding=utf-8
--conf spark.executor.extraJavaOptions=-Dfile.encoding=utf-8

调用算法接口(推荐)

lib库中提供了 10 种常用图计算算法,用户可以通过编程调用的形式调用算法。

  1. 在文件pom.xml中添加依赖。

    <dependency>
         <groupId>com.vesoft</groupId>
         <artifactId>nebula-algorithm</artifactId>
         <version>3.2.0</version>
    </dependency>
    
  2. 传入参数调用算法(以 PageRank 为例)。更多调用示例请参见示例

    Note

    执行算法的 DataFrame 默认第一列是起始点,第二列是目的点,第三列是边权重(非 NebulaGraph 中的 Rank)。

    val prConfig = new PRConfig(5, 1.0)
    val prResult = PageRankAlgo.apply(spark, data, prConfig, false)
    

    如果用户的节点 ID 是 String 类型,可以参考 PageRank 的示例。示例中进行了 ID 转换,将 String 类型编码为 Long 类型, 并在算法结果中将 Long 类型 ID 解码为原始的 String 类型。

直接提交算法包

  1. 设置配置文件

    {
        # 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、graphtriangleCount。
        executeAlgo: pagerank
    
        # PageRank 参数
        pagerank: {
            maxIter: 10
            resetProb: 0.15  
            encodeId:false # 如果点 ID 是字符串类型,请配置为 true。
        }
    
        # Louvain 参数
        louvain: {
            maxIter: 20
            internalIter: 10
            tol: 0.5
            encodeId:false # 如果点 ID 是字符串类型,请配置为 true。
        }
    
        # ...
        }
        }
    }
    

    Note

    当配置为 sink: nebula 的时候,意味着算法运算结果将被写回 NebulaGraph 集群,这对写回到的 TAG 中的属性名有隐含的约定。详情参考本文支持算法部分。

  2. 提交图计算任务。

    ${SPARK_HOME}/bin/spark-submit --master <mode> --class com.vesoft.nebula.algorithm.Main <nebula-algorithm-3.2.0.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-3.0-SNAPSHOT.jar -p /root/nebula-algorithm/src/main/resources/application.conf
    

视频


最后更新: September 6, 2024