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

@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(&#34;Only support one arguments, usage: func(sourceId)&#34;);
    }
    this.sourceId = ((Number) params[0]).longValue();
}

@Override
public void process(RowVertex vertex, Iterator&lt;Long&gt; messages) {
    List&lt;RowEdge&gt; 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)) &lt; 1) {
                Long meg = messages.next();
                context.sendMessage(sourceId, meg);
                // 向邻接点发送与目标点距离
                sendMessageToNeighbors(edges, meg + 1);
                // 标记该点已向目标点发送过消息
                context.updateVertexValue(ObjectRow.create(0L, 1L));
            }
        }
    }
}

private void sendMessageToNeighbors(List&lt;RowEdge&gt; 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(&#34;id&#34;, LongType.INSTANCE, false),
        new TableField(&#34;cc&#34;, DoubleType.INSTANCE, false)
    );
}

}

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)

;

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)

;