How to implement the single source shortest path algorithm - MaxCompute

L'algorithme du plus court chemin à source unique (SSSP) calcule le chemin le plus court entre un sommet source spécifié et tous les autres sommets accessibles dans un graphe. L'algorithme de Dijkstra est une méthode classique pour résoudre le problème SSSP dans un graphe orienté.

Fonctionnement

L'algorithme de Dijkstra utilise des vertices pour mettre à jour les shortest distance values. Chaque vertex conserve sa shortest distance value actuelle par rapport au source vertex. Lorsque cette valeur change, le sommet ajoute le weight d'une edge à la nouvelle valeur et envoie un message pour notifier ses adjacent vertices. Lors de l'itération suivante, les adjacent vertices mettent à jour leurs shortest distance values actuelles en fonction des messages reçus. L'itération prend fin lorsque les shortest distance values actuelles de tous les vertices ne changent plus.

Initialisation : la distance du sommet source s vers lui-même est de 0 (d[s]=0), et la distance de tout autre sommet u vers s est infinie (d[u]=∞).
Itération : si une arête existe de u vers v, la distance minimale de s vers v est mise à jour selon la formule d[v]=min(d[v], d[u]+weight(u, v)). Le processus se poursuit jusqu'à ce que les distances de s vers tous les autres sommets se stabilisent.

Remarque

Pour un graphe orienté pondéré G=(V,E), plusieurs chemins peuvent exister entre un sommet source s et un sommet puits v. Le chemin dont la somme des poids des arêtes est minimale constitue le plus court chemin de s vers v.

Cet algorithme s'adapte naturellement au modèle de programmation MaxCompute Graph.

Cas d'utilisation

MaxCompute prend en charge les graphes orientés et non orientés. Étant donné que la disponibilité des chemins dépend des données sources et de la construction du graphe, les résultats SSSP peuvent différer selon le type de graphe. Le graphe orienté constitue le modèle de données fondamental sur lequel reposent tous les calculs.

Exemples de code

Les exemples suivants illustrent des implémentations SSSP pour les graphes orientés et non orientés.

Graphe orienté

Définissez la classe BaseLoadingVertexResolver. Cette classe est référencée dans la classe principale SSSP.

import com.aliyun.odps.graph.Edge;
import com.aliyun.odps.graph.LoadingVertexResolver;
import com.aliyun.odps.graph.Vertex;
import com.aliyun.odps.graph.VertexChanges;
import com.aliyun.odps.io.Writable;
import com.aliyun.odps.io.WritableComparable;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

@SuppressWarnings("rawtypes")
public class BaseLoadingVertexResolver<I extends WritableComparable, V extends Writable, E extends Writable, M extends Writable>
        extends LoadingVertexResolver<I, V, E, M> {
  @Override
  public Vertex<I, V, E, M> resolve(I vertexId, VertexChanges<I, V, E, M> vertexChanges) throws IOException {

    Vertex<I, V, E, M> vertex = addVertexIfDesired(vertexId, vertexChanges);

    if (vertex != null) {
      addEdges(vertex, vertexChanges);
    } else {
      System.err.println("Ignore all addEdgeRequests for vertex#" + vertexId);
    }
    return vertex;
  }

  protected Vertex<I, V, E, M> addVertexIfDesired(
          I vertexId,
          VertexChanges<I, V, E, M> vertexChanges) {
    Vertex<I, V, E, M> vertex = null;
    if (hasVertexAdditions(vertexChanges)) {
      vertex = vertexChanges.getAddedVertexList().get(0);
    }

    return vertex;
  }

  protected void addEdges(Vertex<I, V, E, M> vertex,
                          VertexChanges<I, V, E, M> vertexChanges) throws IOException {
    Set<I> destVertexId = new HashSet<I>();
    if (vertex.hasEdges()) {
      List<Edge<I, E>> edgeList = vertex.getEdges();
      for (Iterator<Edge<I, E>> edges = edgeList.iterator(); edges.hasNext(); ) {
        Edge<I, E> edge = edges.next();
        if (destVertexId.contains(edge.getDestVertexId())) {
          edges.remove();
        } else {
          destVertexId.add(edge.getDestVertexId());
        }
      }
    }

    for (Vertex<I, V, E, M> vertex1 : vertexChanges.getAddedVertexList()) {
      if (vertex1.hasEdges()) {
        List<Edge<I, E>> edgeList = vertex1.getEdges();
        for (Edge<I, E> edge : edgeList) {
          if (destVertexId.contains(edge.getDestVertexId())) continue;
          destVertexId.add(edge.getDestVertexId());
          vertex.addEdge(edge.getDestVertexId(), edge.getValue());
        }
      }
    }
  }

  protected boolean hasVertexAdditions(VertexChanges<I, V, E, M> changes) {
    return changes != null && changes.getAddedVertexList() != null
            && !changes.getAddedVertexList().isEmpty();
  }
}

Explication du code :

Ligne 15 : définit BaseLoadingVertexResolver. Cette classe gère les conflits survenant lors du chargement des données pour un graphe orienté.
Ligne 18 : la méthode resolve contient la logique de gestion des conflits. Par exemple, si un sommet est ajouté deux fois via deux opérations addVertexRequest, un conflit de chargement se produit. Vous devez résoudre ce conflit avant que le calcul ne puisse se poursuivre.

Définissez la classe SSSP.

import java.io.IOException;

import com.aliyun.odps.graph.Combiner;
import com.aliyun.odps.graph.ComputeContext;
import com.aliyun.odps.graph.Edge;
import com.aliyun.odps.graph.GraphJob;
import com.aliyun.odps.graph.GraphLoader;
import com.aliyun.odps.graph.MutationContext;
import com.aliyun.odps.graph.Vertex;
import com.aliyun.odps.graph.WorkerContext;
import com.aliyun.odps.io.WritableRecord;
import com.aliyun.odps.io.LongWritable;
import com.aliyun.odps.data.TableInfo;

public class SSSP {
  public static final String START_VERTEX = "sssp.start.vertex.id";

  public static class SSSPVertex extends
          Vertex<LongWritable, LongWritable, LongWritable, LongWritable> {
    private static long startVertexId = -1;

    public SSSPVertex() {
      this.setValue(new LongWritable(Long.MAX_VALUE));
    }

    public boolean isStartVertex(
            ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context) {
      if (startVertexId == -1) {
        String s = context.getConfiguration().get(START_VERTEX);
        startVertexId = Long.parseLong(s);
      }
      return getId().get() == startVertexId;
    }

    @Override
    public void compute(
            ComputeContext<LongWritable, LongWritable, LongWritable, LongWritable> context,
            Iterable<LongWritable> messages) throws IOException {
      long minDist = isStartVertex(context) ? 0 : Long.MAX_VALUE;
      for (LongWritable msg : messages) {
        if (msg.get() < minDist) {
          minDist = msg.get();
        }
      }
      if (minDist < this.getValue().get()) {
        this.setValue(new LongWritable(minDist));
        if (hasEdges()) {
          for (Edge<LongWritable, LongWritable> e : this.getEdges()) {
            context.sendMessage(e.getDestVertexId(), new LongWritable(minDist + e.getValue().get()));
          }
        }
      } else {
        voteToHalt();
      }
    }

    @Override
    public void cleanup(
            WorkerContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
            throws IOException {
      context.write(getId(), getValue());
    }

    @Override
    public String toString() {
      return "Vertex(id=" + this.getId() + ",value=" + this.getValue() + ",#edges=" + this.getEdges() + ")";
    }
  }

  public static class SSSPGraphLoader extends
          GraphLoader<LongWritable, LongWritable, LongWritable, LongWritable> {
    @Override
    public void load(
            LongWritable recordNum,
            WritableRecord record,
            MutationContext<LongWritable, LongWritable, LongWritable, LongWritable> context)
            throws IOException {
      SSSPVertex vertex = new SSSPVertex();
      vertex.setId((LongWritable) record.get(0));
      String[] edges = record.get(1).toString().split(",");
      for (String edge : edges) {
        String[] ss = edge.split(":");
        vertex.addEdge(new LongWritable(Long.parseLong(ss[0])), new LongWritable(Long.parseLong(ss[1])));
      }
      context.addVertexRequest(vertex);
    }
  }

  public static class MinLongCombiner extends
          Combiner<LongWritable, LongWritable> {
    @Override
    public void combine(LongWritable vertexId, LongWritable combinedMessage,
                        LongWritable messageToCombine) throws IOException {
      if (combinedMessage.get() > messageToCombine.get()) {
        combinedMessage.set(messageToCombine.get());
      }
    }
  }

  public static void main(String[] args) throws IOException {
    if (args.length < 3) {
      System.out.println("Usage: <startnode> <input> <output>");
      System.exit(-1);
    }
    GraphJob job = new GraphJob();
    job.setGraphLoaderClass(SSSPGraphLoader.class);
    job.setVertexClass(SSSPVertex.class);
    job.setCombinerClass(MinLongCombiner.class);
    job.setLoadingVertexResolver(BaseLoadingVertexResolver.class);
    job.set(START_VERTEX, args[0]);
    job.addInput(TableInfo.builder().tableName(args[1]).build());
    job.addOutput(TableInfo.builder().tableName(args[2]).build());
    long startTime = System.currentTimeMillis();
    job.run();
    System.out.println("Job Finished in "
            + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
  }
}

Explication du code :

Ligne 19 : définit SSSPVertex. Dans cette classe :
- La valeur du sommet représente la distance minimale entre ce sommet et le sommet source startVertexId.
- La méthode compute() utilise la formule itérative d[v]=min(d[v], d[u]+weight(u, v)) pour calculer la distance minimale et mettre à jour la valeur du sommet actuel.
- La méthode cleanup() écrit la distance minimale entre le sommet actuel et le sommet source dans la table de sortie.
Ligne 54 : si la Value du sommet actuel (le plus court chemin entre le sommet et le sommet source) ne change pas, appelez la méthode voteToHalt() via le framework pour faire passer le sommet à l'état halt. Lorsque tous les sommets entrent dans l'état halt, le calcul se termine.
Ligne 71 : définit un GraphLoader pour charger les données du graphe en tant que graphe orienté. Il analyse les enregistrements de la table en sommets et arêtes de graphe et les charge dans le framework. Dans cet exemple, la méthode addVertexRequest charge les informations des sommets dans le contexte de calcul du graphe.
Ligne 90 : définit MinLongCombiner. Celle-ci combine les messages envoyés au même sommet afin d'optimiser les performances et de réduire la consommation de mémoire.
Ligne 101 : la fonction main définit le GraphJob. Elle configure les implémentations pour Vertex, GraphLoader, BaseLoadingVertexResolver et Combiner, ainsi que les tables d'entrée et de sortie.
Ligne 110 : définit la classe BaseLoadingVertexResolver pour gérer les conflits.

Graphe non orienté

import com.aliyun.odps.data.TableInfo;
import com.aliyun.odps.graph.*;
import com.aliyun.odps.io.DoubleWritable;
import com.aliyun.odps.io.LongWritable;
import com.aliyun.odps.io.WritableRecord;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

public class SSSPBenchmark4 {
    public static final String START_VERTEX = "sssp.start.vertex.id";

    public static class SSSPVertex extends
            Vertex<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> {
        private static long startVertexId = -1;
        public SSSPVertex() {
            this.setValue(new DoubleWritable(Double.MAX_VALUE));
        }
        public boolean isStartVertex(
                ComputeContext<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> context) {
            if (startVertexId == -1) {
                String s = context.getConfiguration().get(START_VERTEX);
                startVertexId = Long.parseLong(s);
            }
            return getId().get() == startVertexId;
        }

        @Override
        public void compute(
                ComputeContext<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> context,
                Iterable<DoubleWritable> messages) throws IOException {
            double minDist = isStartVertex(context) ? 0 : Double.MAX_VALUE;
            for (DoubleWritable msg : messages) {
                if (msg.get() < minDist) {
                    minDist = msg.get();
                }
            }
            if (minDist < this.getValue().get()) {
                this.setValue(new DoubleWritable(minDist));
                if (hasEdges()) {
                    for (Edge<LongWritable, DoubleWritable> e : this.getEdges()) {
                        context.sendMessage(e.getDestVertexId(), new DoubleWritable(minDist
                                + e.getValue().get()));
                    }
                }
            } else {
                voteToHalt();
            }
        }

        @Override
        public void cleanup(
                WorkerContext<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> context)
                throws IOException {
            context.write(getId(), getValue());
        }
    }

    public static class MinLongCombiner extends
            Combiner<LongWritable, DoubleWritable> {
        @Override
        public void combine(LongWritable vertexId, DoubleWritable combinedMessage,
                            DoubleWritable messageToCombine) {
            if (combinedMessage.get() > messageToCombine.get()) {
                combinedMessage.set(messageToCombine.get());
            }
        }
    }

    public static class SSSPGraphLoader extends
            GraphLoader<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> {
        @Override
        public void load(
                LongWritable recordNum,
                WritableRecord record,
                MutationContext<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> context)
                throws IOException {
            LongWritable sourceVertexID = (LongWritable) record.get(0);
            LongWritable destinationVertexID = (LongWritable) record.get(1);
            DoubleWritable edgeValue = (DoubleWritable) record.get(2);
            Edge<LongWritable, DoubleWritable> edge = new Edge<LongWritable, DoubleWritable>(destinationVertexID, edgeValue);
            context.addEdgeRequest(sourceVertexID, edge);
            Edge<LongWritable, DoubleWritable> edge2 = new
                    Edge<LongWritable, DoubleWritable>(sourceVertexID, edgeValue);
            context.addEdgeRequest(destinationVertexID, edge2);
        }
    }

    public static class SSSPLoadingVertexResolver extends
            LoadingVertexResolver<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> {

        @Override
        public Vertex<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> resolve(
                LongWritable vertexId,
                VertexChanges<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> vertexChanges) throws IOException {

            SSSPVertex computeVertex = new SSSPVertex();
            computeVertex.setId(vertexId);
            Set<LongWritable> destinationVertexIDSet = new HashSet<>();

            if (hasEdgeAdditions(vertexChanges)) {
                for (Edge<LongWritable, DoubleWritable> edge : vertexChanges.getAddedEdgeList()) {
                    if (!destinationVertexIDSet.contains(edge.getDestVertexId())) {
                        destinationVertexIDSet.add(edge.getDestVertexId());
                        computeVertex.addEdge(edge.getDestVertexId(), edge.getValue());
                    }

                }
            }

            return computeVertex;
        }

        protected boolean hasEdgeAdditions(VertexChanges<LongWritable, DoubleWritable, DoubleWritable, DoubleWritable> changes) {
            return changes != null && changes.getAddedEdgeList() != null
                    && !changes.getAddedEdgeList().isEmpty();
        }
    }

    public static void main(String[] args) throws IOException {
        if (args.length < 2) {
            System.out.println("Usage: <startnode> <input> <output>");
            System.exit(-1);
        }
        GraphJob job = new GraphJob();
        job.setGraphLoaderClass(SSSPGraphLoader.class);
        job.setLoadingVertexResolver(SSSPLoadingVertexResolver.class);
        job.setVertexClass(SSSPVertex.class);
        job.setCombinerClass(MinLongCombiner.class);
        job.set(START_VERTEX, args[0]);
        job.addInput(TableInfo.builder().tableName(args[1]).build());
        job.addOutput(TableInfo.builder().tableName(args[2]).build());
        long startTime = System.currentTimeMillis();
        job.run();
        System.out.println("Job Finished in "
                + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
    }
}

Explication du code :

Ligne 15 : définit SSSPVertex. Dans cette classe :
- La valeur du sommet représente la distance minimale entre ce sommet et le sommet source startVertexId.
- La méthode compute() utilise la formule itérative d[v]=min(d[v], d[u]+weight(u, v)) pour calculer la distance minimale et mettre à jour la valeur du sommet actuel.
- La méthode cleanup() écrit la distance minimale entre le sommet actuel et le sommet source dans la table de sortie.
Ligne 54 : si la Value du sommet actuel (le plus court chemin entre ce sommet et le sommet source) ne change pas, le sommet passe à l'état halt en appelant voteToHalt() via le framework. Le calcul se termine lorsque tous les sommets entrent dans l'état halt.
Ligne 61 : définit MinLongCombiner. Celle-ci combine les messages envoyés au même sommet afin d'optimiser les performances et de réduire la consommation de mémoire.
Ligne 72 : définit un GraphLoader pour charger les données du graphe en tant que graphe non orienté. Il utilise addEdgeRequest pour charger l'arête entre deux sommets comme une arête bidirectionnelle, garantissant ainsi que les données de la table sont chargées sous forme de graphe non orienté.
- Ligne 80 : la première colonne représente l'ID du sommet source.
- Ligne 81 : la deuxième colonne représente l'ID du sommet de destination.
- Ligne 82 : la troisième colonne représente le poids de l'arête.
- Ligne 83 : crée une arête composée de l'ID du sommet de destination et du poids de l'arête.
- Ligne 84 : demande l'ajout de l'arête au sommet source.
- Lignes 85-87 : chaque Record représente une arête bidirectionnelle. La logique des lignes 83 et 84 est répétée pour la direction inverse.
Définit SSSPLoadingVertexResolver. Cette classe gère les conflits survenant lors du chargement des données pour un graphe non orienté. Par exemple, si la même arête est ajoutée deux fois via deux opérations addEdgeRequest, un conflit de chargement se produit. Vous devez gérer les arêtes en double pour garantir l'exactitude du calcul.
Ligne 101 : la fonction main définit le GraphJob. Elle configure les implémentations pour Vertex, GraphLoader, SSSPLoadingVertexResolver et Combiner, ainsi que les tables d'entrée et de sortie.

Résultats d'exécution

La sortie suivante provient de l'exécution de l'exemple de code pour le graphe orienté. Pour plus d'informations, consultez Développer des programmes Graph.

vertex    value
1        0
2        2
3        1
4        3
5        2

vertex : le sommet actuel.
value : la distance minimale entre le sommet actuel et le sommet source (1).

Remarque

Pour créer des données pour un graphe non orienté, utilisez l'ID du sommet source, l'ID du sommet de destination et le poids de l'arête, comme illustré dans l'exemple de code précédent.

Tutoriel

Pour plus d'informations sur la mise en œuvre des exemples de code précédents, consultez Développer des programmes Graph.