Tugraph Analytics图计算快速上手之紧密中心度算法

原创

GeaFlow

发布于 2023-09-19 14:51:40

4200

发布于 2023-09-19 14:51:40

文章被收录于专栏：流图计算

作者：张武科

概述

紧密中心度（Closeness Centrality）计量了一个节点到其他所有节点的紧密性，即该节点到其他节点的距离的倒数；节点对应的值越高表示紧密性越好，能够在图中传播信息的能力越强，可用以衡量信息流入或流出该节点的能力，多用与社交网络中关键节点发掘等场景。

算法介绍

对于图中一个给定节点，紧密性中心性是该节点到其他所有节点的最小距离和的倒数：

其中，u表示待计算紧密中心度的节点，d(u, v)表示节点u到节点v的最短路径距离；实际场景中，更多地使用归一化后的紧密中心度，即计算目标节点到其他节点的平均距离的倒数：

其中，n表示图中节点数。

算法实现

首先，基于AlgorithmUserFunction接口实现ClosenessCentrality，如下：

@Description(name = "closeness_centrality", description = "built-in udga for ClosenessCentrality")
public class ClosenessCentrality implements AlgorithmUserFunction<Long, Long> {

    private AlgorithmRuntimeContext context;
    private long sourceId;

    @Override
    public void init(AlgorithmRuntimeContext context, Object[] params) {
        this.context = context;
        if (params.length != 1) {
            throw new IllegalArgumentException("Only support one arguments, usage: func(sourceId)");
        }
        this.sourceId = ((Number) params[0]).longValue();
    }

    @Override
    public void process(RowVertex vertex, Iterator<Long> messages) {
        List<RowEdge> edges = context.loadEdges(EdgeDirection.OUT);
        if (context.getCurrentIterationId() == 1L) {
            context.sendMessage(vertex.getId(), 1L);
            context.sendMessage(sourceId, 1L);
        } else if (context.getCurrentIterationId() == 2L) {
            context.updateVertexValue(ObjectRow.create(0L, 0L));
            if (vertex.getId().equals(sourceId)) {
                long vertexNum = -2L;
                while (messages.hasNext()) {
                    messages.next();
                    vertexNum++;
                }
                // 统计节点数
                context.updateVertexValue(ObjectRow.create(0L, vertexNum));
                // 向邻接点发送与目标点距离
                sendMessageToNeighbors(edges, 1L);
            }
        } else {
            if (vertex.getId().equals(sourceId)) {
                long sum = (long) vertex.getValue().getField(0, LongType.INSTANCE);
                while (messages.hasNext()) {
                    sum += messages.next();
                }
                long vertexNum = (long) vertex.getValue().getField(1, LongType.INSTANCE);
                // 记录最短距离和
                context.updateVertexValue(ObjectRow.create(sum, vertexNum));
            } else {
                if (((long) vertex.getValue().getField(1, LongType.INSTANCE)) < 1) {
                    Long meg = messages.next();
                    context.sendMessage(sourceId, meg);
                    // 向邻接点发送与目标点距离
                    sendMessageToNeighbors(edges, meg + 1);
                    // 标记该点已向目标点发送过消息
                    context.updateVertexValue(ObjectRow.create(0L, 1L));
                }
            }
        }
    }

    private void sendMessageToNeighbors(List<RowEdge> outEdges, Object message) {
        for (RowEdge rowEdge : outEdges) {
            context.sendMessage(rowEdge.getTargetId(), message);
        }
    }

    @Override
    public void finish(RowVertex vertex) {
        if (vertex.getId().equals(sourceId)) {
            long len = (long) vertex.getValue().getField(0, LongType.INSTANCE);
            long num = (long) vertex.getValue().getField(1, LongType.INSTANCE);
            context.take(ObjectRow.create(vertex.getId(), (double) num / len));
        }
    }

    @Override
    public StructType getOutputType() {
        return new StructType(
            new TableField("id", LongType.INSTANCE, false),
            new TableField("cc", DoubleType.INSTANCE, false)
        );
    }
}

然后，可在 DSL 中引入自定义算法，直接调用使用，语法形式如下：

CREATE Function closeness_centrality AS 'com.antgroup.geaflow.dsl.udf.ClosenessCentrality';

INSERT INTO tbl_result
CALL closeness_centrality(1) YIELD (vid, ccValue)
RETURN vid, ROUND(ccValue, 3)
;

示例表示，计算图中 id = 1节点的紧密中心度。

算法运行

在运行算法之前，要构造算法运行的底图数据。

图定义

首先，进行图定义：

CREATE GRAPH modern (
	Vertex person (
	  id bigint ID,
	  name varchar,
	  age int
	),
	Vertex software (
	  id bigint ID,
	  name varchar,
	  lang varchar
	),
	Edge knows (
	  srcId bigint SOURCE ID,
	  targetId bigint DESTINATION ID,
	  weight double
	),
	Edge created (
	  srcId bigint SOURCE ID,
  	targetId bigint DESTINATION ID,
  	weight double
	)
) WITH (
	storeType='rocksdb',
	shardNum = 1
);

图构建

完成图定义之后，导入点边数据，构造数据底图：

CREATE TABLE modern_vertex (
  id varchar,
  type varchar,
  name varchar,
  other varchar
) WITH (
  type='file',
  geaflow.dsl.file.path = 'resource:///data/modern_vertex.txt'
);

CREATE TABLE modern_edge (
  srcId bigint,
  targetId bigint,
  type varchar,
  weight double
) WITH (
  type='file',
  geaflow.dsl.file.path = 'resource:///data/modern_edge.txt'
);

INSERT INTO modern.person
SELECT cast(id as bigint), name, cast(other as int) as age
FROM modern_vertex WHERE type = 'person'
;

INSERT INTO modern.software
SELECT cast(id as bigint), name, cast(other as varchar) as lang
FROM modern_vertex WHERE type = 'software'
;

INSERT INTO modern.knows
SELECT srcId, targetId, weight
FROM modern_edge WHERE type = 'knows'
;

INSERT INTO modern.created
SELECT srcId, targetId, weight
FROM modern_edge WHERE type = 'created'
;

计算输出

最后，在底图数据上完成算法计算和结果输出；

CREATE TABLE tbl_result (
  vid int,
	ccValue double
) WITH (
	type='file',
	geaflow.dsl.file.path='/tmp/result'
);

CREATE Function closeness_centrality AS 'com.antgroup.geaflow.dsl.udf.ClosenessCentrality';

USE GRAPH modern;

INSERT INTO tbl_result
CALL closeness_centrality(1) YIELD (vid, ccValue)
RETURN vid, ROUND(ccValue, 3)
;

运行示例

input// vertex 1,person,marko,29 2,person,vadas,27 3,software,lop,java 4,person,josh,32 5,software,ripple,java 6,person,peter,35 // edge 1,3,created,0.4 1,2,knows,0.5 1,4,knows,1.0 4,3,created,0.4 4,5,created,1.0 3,6,created,0.2
output// result 1,0.714