flink 实现三角枚举EnumTriangles算法详解

/**
 * @Author: xu.dm
 * @Date: 2019/7/4 21:31
 * @Description: 三角枚举算法
 * 三角枚举是在图（数据结构）中找到紧密连接的部分的预处理步骤。三角形由三条边连接，三条边相互连接。
 *
 * 该算法的工作原理如下：它将所有共享一个共同顶点的边(edge)分组，并构建三元组，即由两条边连接的顶点三元组。
 * 最后，通过join操作连接原数据和三元组，过滤所有三元组，去除不存在第三条边(闭合三角形)的三元组。
 *
 * 输入文件是纯文本文件，必须格式如下：
 *
 * 边缘表示为顶点ID的对，由空格字符分隔。边线由换行符分隔。
 * 例如，"1 2\n2 12\n1 12\n42 63"给出包括三角形的四个（无向）边（1） - （2），（2） - （12），（1） - （12）和（42） - （63）
 *     （1）
 *     /    *  （2）-（12）
 * 用法：EnumTriangleBasic --edges <path> --output <path>
 * 如果未提供参数，则使用{@link EnumTrianglesData}中的默认数据运行程序。
 *
 */
public class EnumTriangles {
    public static void main(String args[]) throws Exception{
        // Checking input parameters
        final ParameterTool params = ParameterTool.fromArgs(args);

        // set up execution environment
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // make parameters available in the web interface
        env.getConfig().setGlobalJobParameters(params);

        // read input data
        DataSet<EnumTrianglesDataTypes.Edge> edges;
        if (params.has("edges")) {
            edges = env.readCsvFile(params.get("edges"))
                    .fieldDelimiter(" ")
                    .includeFields(true, true)
                    .types(Integer.class, Integer.class)
                    .map(new TupleEdgeConverter());
        } else {
            System.out.println("Executing EnumTriangles example with default edges data set.");
            System.out.println("Use --edges to specify file input.");
            edges = EnumTrianglesData.getDefaultEdgeDataSet(env);
        }

        // project edges by vertex id
        // 对Edge里V1,V2排序，让小的在前
        DataSet<EnumTrianglesDataTypes.Edge> edgesById = edges
                .map(new EdgeByIdProjector());

        DataSet<EnumTrianglesDataTypes.Triad> triangles = edgesById
                // build triads
                .groupBy(EnumTrianglesDataTypes.Edge.V1)
                .sortGroup(EnumTrianglesDataTypes.Edge.V2, Order.ASCENDING)
                .reduceGroup(new TriadBuilder())
                //通过Triad的第二和第三字段与Edge的第一和第二字段join，找出有第二和第三有路径的对子。
                .join(edgesById)
                .where(EnumTrianglesDataTypes.Triad.V2, EnumTrianglesDataTypes.Triad.V3)
                .equalTo(EnumTrianglesDataTypes.Edge.V1, EnumTrianglesDataTypes.Edge.V2)
                // filter triads
                .with(new TriadFilter());

        // emit result
        if (params.has("output")) {
            triangles.writeAsCsv(params.get("output"), "\n", ",");
            // execute program
            env.execute("Basic Triangle Enumeration Example");
        } else {
            System.out.println("Printing result to stdout. Use --output to specify output path.");
            triangles.print();
        }
    }

    //转换Tuple2类型到Edge类型
    @FunctionAnnotation.ForwardedFields("0;1")
    public static class TupleEdgeConverter implements MapFunction<Tuple2<Integer,Integer>, EnumTrianglesDataTypes.Edge>{
        private final EnumTrianglesDataTypes.Edge outEdge = new EnumTrianglesDataTypes.Edge();
        @Override
        public EnumTrianglesDataTypes.Edge map(Tuple2<Integer, Integer> value) throws Exception {
            outEdge.copyVerticesFromTuple2(value);
            return outEdge;
        }
    }

    /**
     * Projects an edge (pair of vertices) such that the id of the first is smaller than the id of the second.
     * 转换一对边，如果first大于second，则互相交换
     * */
    private static class EdgeByIdProjector implements MapFunction<EnumTrianglesDataTypes.Edge, EnumTrianglesDataTypes.Edge>{
        @Override
        public EnumTrianglesDataTypes.Edge map(EnumTrianglesDataTypes.Edge value) throws Exception {

            //flip vertices 反转顶点
            if(value.getFirstVertex()>value.getSecondVertex())
            {
                value.flipVertices();
            }
            return value;
        }
    }

    /**
     *  Builds triads (triples of vertices) from pairs of edges that share a vertex.
     *  The first vertex of a triad is the shared vertex, the second and third vertex are ordered by vertexId.
     *  Assumes that input edges share the first vertex and are in ascending order of the second vertex.
     *  构造triads（三位一体结构），即经过分组排序后，拥有共同顶点的edge，按顺序（分组后顶点需要升序）组合成三体
     *  构成Triad的第一个顶点是分组id，也是分组里所有对子（edge）共享的id，第二和第三个顶点是分组的对子（edge）里的第二个顶点，edge他们是按升序排列。
     *  分组结构类似如下，s1就是共享的顶点，f1到f3是按升序排列的对子(s1-f1),(s1-f2),(f1-f3)。
     *       f1
     *     /
     *  s1 一 f2
     *          *     f3
     */
    @FunctionAnnotation.ForwardedFields("0")
    private static class TriadBuilder
            implements GroupReduceFunction<EnumTrianglesDataTypes.Edge, EnumTrianglesDataTypes.Triad>{
        private final List<Integer> vertices = new ArrayList<>();
        private final EnumTrianglesDataTypes.Triad outTriad = new EnumTrianglesDataTypes.Triad();

        @Override
        public void reduce(Iterable<EnumTrianglesDataTypes.Edge> values, Collector<EnumTrianglesDataTypes.Triad> out) throws Exception {
            final Iterator<EnumTrianglesDataTypes.Edge> edges = values.iterator();

            // clear vertex list
            vertices.clear();

            // read first edge
            EnumTrianglesDataTypes.Edge firstEdge = edges.next();
            outTriad.setFirstVertex(firstEdge.getFirstVertex());

            vertices.add(firstEdge.getSecondVertex());

            // build and emit triads
            if(edges.hasNext()){
                Integer higherVertexId = edges.next().getSecondVertex();

                //combine vertex with all previously read vertices
                for(Integer lowVertexId:vertices){
                    outTriad.setSecondVertex(lowVertexId);
                    outTriad.setThirdVertex(higherVertexId);
                    out.collect(outTriad);
                }
                vertices.add(higherVertexId);
            }
        }
    }

    /**
     * Filters triads (three vertices connected by two edges) without a closing third edge.
     * 过滤没有闭合边的三体
     * */
    private static class TriadFilter
            implements JoinFunction<EnumTrianglesDataTypes.Triad, EnumTrianglesDataTypes.Edge, EnumTrianglesDataTypes.Triad>{
        @Override
        public EnumTrianglesDataTypes.Triad join(EnumTrianglesDataTypes.Triad first, EnumTrianglesDataTypes.Edge second) throws Exception {
            return first;
        }
    }

}

public class EnumTrianglesDataTypes {
    /**
     * A POJO storing two vertex IDs.
     */
    public static class Edge extends Tuple2<Integer, Integer> {
        private static final long serialVersionUID = 1L;

        public static final int V1 = 0;
        public static final int V2 = 1;

        public Edge() {}

        public Edge(final Integer v1, final Integer v2) {
            this.setFirstVertex(v1);
            this.setSecondVertex(v2);
        }

        public Integer getFirstVertex() {
            return this.getField(V1);
        }

        public Integer getSecondVertex() {
            return this.getField(V2);
        }

        public void setFirstVertex(final Integer vertex1) {
            this.setField(vertex1, V1);
        }

        public void setSecondVertex(final Integer vertex2) {
            this.setField(vertex2, V2);
        }

        public void copyVerticesFromTuple2(Tuple2<Integer, Integer> t) {
            this.setFirstVertex(t.f0);
            this.setSecondVertex(t.f1);
        }

        public void copyVerticesFromEdgeWithDegrees(EdgeWithDegrees ewd) {
            this.setFirstVertex(ewd.getFirstVertex());
            this.setSecondVertex(ewd.getSecondVertex());
        }

        public void flipVertices() {
            Integer tmp = this.getFirstVertex();
            this.setFirstVertex(this.getSecondVertex());
            this.setSecondVertex(tmp);
        }
    }

    /**
     * A POJO storing three vertex IDs.
     */
    public static class Triad extends Tuple3<Integer, Integer, Integer> {
        private static final long serialVersionUID = 1L;

        public static final int V1 = 0;
        public static final int V2 = 1;
        public static final int V3 = 2;

        public Triad() {}

        public void setFirstVertex(final Integer vertex1) {
            this.setField(vertex1, V1);
        }

        public void setSecondVertex(final Integer vertex2) {
            this.setField(vertex2, V2);
        }

        public void setThirdVertex(final Integer vertex3) {
            this.setField(vertex3, V3);
        }
    }

    /**
     * A POJO storing two vertex IDs with degree.
     */
    public static class EdgeWithDegrees extends Tuple4<Integer, Integer, Integer, Integer> {
        private static final long serialVersionUID = 1L;

        public static final int V1 = 0;
        public static final int V2 = 1;
        public static final int D1 = 2;
        public static final int D2 = 3;

        public EdgeWithDegrees() { }

        public Integer getFirstVertex() {
            return this.getField(V1);
        }

        public Integer getSecondVertex() {
            return this.getField(V2);
        }

        public Integer getFirstDegree() {
            return this.getField(D1);
        }

        public Integer getSecondDegree() {
            return this.getField(D2);
        }

        public void setFirstVertex(final Integer vertex1) {
            this.setField(vertex1, V1);
        }

        public void setSecondVertex(final Integer vertex2) {
            this.setField(vertex2, V2);
        }

        public void setFirstDegree(final Integer degree1) {
            this.setField(degree1, D1);
        }

        public void setSecondDegree(final Integer degree2) {
            this.setField(degree2, D2);
        }

        public void copyFrom(final EdgeWithDegrees edge) {
            this.setFirstVertex(edge.getFirstVertex());
            this.setSecondVertex(edge.getSecondVertex());
            this.setFirstDegree(edge.getFirstDegree());
            this.setSecondDegree(edge.getSecondDegree());
        }
    }
}

public class EnumTrianglesData {
    public static final Object[][] EDGES = {
            {1, 2},
            {1, 3},
            {1, 4},
            {1, 5},
            {2, 3},
            {2, 5},
            {3, 4},
            {3, 5},
            {3, 7},
            {3, 8},
            {5, 6},
            {7, 8}
    };

    public static DataSet<EnumTrianglesDataTypes.Edge> getDefaultEdgeDataSet(ExecutionEnvironment env) {

        List<EnumTrianglesDataTypes.Edge> edges = new ArrayList<EnumTrianglesDataTypes.Edge>();
        for (Object[] e : EDGES) {
            edges.add(new EnumTrianglesDataTypes.Edge((Integer) e[0], (Integer) e[1]));
        }

        return env.fromCollection(edges);
    }
}