MaxCompute:Ruta más corta desde un solo origen

Funcionamiento

El algoritmo de Dijkstra utiliza vertices para actualizar los shortest distance values. Cada vertex mantiene su shortest distance value actual respecto al source vertex. Cuando este valor cambia, el vértice suma el weight de una edge al nuevo valor y envía un mensaje para notificar a sus adjacent vertices. En la siguiente iteración, los adjacent vertices actualizan sus shortest distance values actuales con base en los mensajes recibidos. La iteración finaliza cuando los shortest distance values actuales de todos los vertices dejan de cambiar.

• Inicialización: La distancia del vértice de origen s a sí mismo es 0 (d[s]=0), y la distancia de cualquier otro vértice u a s es infinito (d[u]=∞).

• Iteración: Si existe una arista de u a v, la distancia más corta de s a v se actualiza como d[v]=min(d[v], d[u]+weight(u, v)). El proceso continúa hasta que las distancias de s a todos los demás vértices se estabilizan.

Este algoritmo se adapta de forma natural al modelo de programación MaxCompute Graph.

Casos de uso

MaxCompute admite tanto grafos dirigidos como no dirigidos. Dado que la disponibilidad de rutas depende de los datos de origen y de la construcción del grafo, los resultados de SSSP pueden variar entre tipos de grafo. El grafo dirigido es el modelo de datos fundamental en el que se basan todos los cálculos.

Ejemplos de código

A continuación se presentan ejemplos que muestran implementaciones de SSSP para grafos dirigidos y no dirigidos.

• Grafo dirigido

  • Defina la clase BaseLoadingVertexResolver. Esta clase se referencia en la clase principal 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();
      }
    }

    Explicación del código:

    • Línea 15: Define BaseLoadingVertexResolver. Esta clase gestiona los conflictos que surgen al cargar datos para un grafo dirigido.

    • Línea 18: El método resolve contiene la lógica para gestionar un conflicto. Por ejemplo, si un vértice se agrega dos veces mediante dos operaciones addVertexRequest, se produce un conflicto de carga. Debe resolver este conflicto antes de que pueda continuar el cálculo.

  • Defina la clase 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");
      }
    }
    
                                

    Explicación del código:

    • Línea 19: Define SSSPVertex. En esta clase:

      • El valor del vértice representa la distancia más corta desde este vértice hasta el vértice de origen startVertexId.

      • El método compute() utiliza la fórmula iterativa d[v]=min(d[v], d[u]+weight(u, v)) para calcular la distancia más corta y actualizar el valor del vértice actual.

      • El método cleanup() escribe la distancia más corta desde el vértice actual hasta el vértice de origen en la tabla de salida.

    • Línea 54: Si el Value del vértice actual (la ruta más corta desde el vértice hasta el vértice de origen) no cambia, llame al método voteToHalt() a través del framework para pasar el vértice al estado halt. Cuando todos los vértices entran en el estado halt, el cálculo finaliza.

    • Línea 71: Define un GraphLoader para cargar los datos del grafo como un grafo dirigido. Analiza sintácticamente los registros de la tabla en vértices y aristas del grafo, y los carga en el framework. En este ejemplo, el método addVertexRequest carga la información de los vértices en el contexto de cálculo del grafo.

    • Línea 90: Define MinLongCombiner. Esto combina los mensajes enviados al mismo vértice para optimizar el rendimiento y reducir el consumo de memoria.

    • Línea 101: La función main define el GraphJob. Establece las implementaciones para Vertex, GraphLoader, BaseLoadingVertexResolver y Combiner, y configura las tablas de entrada y salida.

    • Línea 110: Establece la clase BaseLoadingVertexResolver para gestionar conflictos.

• Grafo no dirigido

  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");
      }
  }
                      

  Explicación del código:

  • Línea 15: Define SSSPVertex. En esta clase:

    • El valor del vértice representa la distancia más corta desde este vértice hasta el vértice de origen startVertexId.

    • El método compute() utiliza la fórmula iterativa d[v]=min(d[v], d[u]+weight(u, v)) para calcular la distancia más corta y actualizar el valor del vértice actual.

    • El método cleanup() escribe la distancia más corta desde el vértice actual hasta el vértice de origen en la tabla de salida.

  • Línea 54: Si el Value del vértice actual (la ruta más corta desde este vértice hasta el vértice de origen) no cambia, el vértice entra en el estado halt al llamar a voteToHalt() a través del framework. El cálculo finaliza cuando todos los vértices entran en el estado halt.

  • Línea 61: Define MinLongCombiner. Esto combina los mensajes enviados al mismo vértice para optimizar el rendimiento y reducir el consumo de memoria.

  • Línea 72: Define un GraphLoader para cargar los datos del grafo como un grafo no dirigido. Utiliza addEdgeRequest para cargar la arista entre dos vértices como una arista bidireccional, lo que garantiza que los datos de la tabla se carguen como un grafo no dirigido.

    • Línea 80: La primera columna representa el ID del vértice de origen.

    • Línea 81: La segunda columna representa el ID del vértice de destino.

    • Línea 82: La tercera columna representa el peso de la arista.

    • Línea 83: Crea una arista compuesta por el ID del vértice de destino y el peso de la arista.

    • Línea 84: Realiza una solicitud para agregar la arista al vértice de origen.

    • Líneas 85-87: Cada Record representa una arista bidireccional. La lógica de las líneas 83 y 84 se repite para la dirección inversa.

  • Define SSSPLoadingVertexResolver. Esta clase gestiona los conflictos que surgen al cargar datos para un grafo no dirigido. Por ejemplo, si la misma arista se agrega dos veces mediante dos operaciones addEdgeRequest, se produce un conflicto de carga. Debe gestionar las aristas duplicadas para garantizar un cálculo correcto.

  • Línea 101: La función main define el GraphJob. Establece las implementaciones para Vertex, GraphLoader, SSSPLoadingVertexResolver y Combiner, y configura las tablas de entrada y salida.

Resultados de ejecución

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

Tutorial

Para obtener más información sobre cómo implementar los ejemplos de código anteriores, consulte Develop Graph programs.