MaxCompute:Single source shortest path

Cara kerja

Algoritma Dijkstra menggunakan vertices untuk memperbarui shortest distance values. Setiap vertex menyimpan shortest distance value saat ini dari source vertex. Ketika nilai ini berubah, vertex tersebut menambahkan weight dari sebuah edge ke nilai tersebut dan mengirim pesan kepada adjacent vertices-nya. Pada iterasi berikutnya, adjacent vertices memperbarui shortest distance values mereka berdasarkan pesan yang diterima. Iterasi berakhir ketika shortest distance values dari semua vertices tidak lagi berubah.

• Inisialisasi: Jarak dari vertex sumber s ke dirinya sendiri adalah 0 (d[s]=0), sedangkan jarak dari vertex lain u ke s diinisialisasi sebagai tak hingga (d[u]=∞).

• Iterasi: Jika terdapat edge dari u ke v, jarak terpendek dari s ke v diperbarui menjadi d[v]=min(d[v], d[u]+weight(u, v)). Proses ini berlanjut hingga jarak dari s ke semua vertex lain stabil.

Algoritma ini secara alami sesuai dengan model pemrograman MaxCompute Graph.

Kasus penggunaan

MaxCompute mendukung graf berarah maupun tak berarah. Karena ketersediaan jalur bergantung pada data sumber dan konstruksi graf, hasil SSSP dapat berbeda antara jenis graf tersebut. Graf berarah merupakan model data dasar tempat semua komputasi didasarkan.

Contoh kode

Contoh berikut menunjukkan implementasi SSSP untuk graf berarah dan tak berarah.

• Graf berarah

  • Definisikan kelas BaseLoadingVertexResolver. Kelas ini dirujuk dalam kelas utama 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();
      }
    }

    Penjelasan kode:

    • Baris 15: Mendefinisikan BaseLoadingVertexResolver. Kelas ini menangani konflik yang terjadi saat memuat data untuk graf berarah.

    • Baris 18: Metode resolve berisi logika untuk menangani konflik. Misalnya, jika sebuah vertex ditambahkan dua kali melalui dua operasi addVertexRequest, terjadi konflik pemuatan. Anda harus menyelesaikan konflik ini sebelum komputasi dapat dilanjutkan.

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

    Penjelasan kode:

    • Baris 19: Mendefinisikan SSSPVertex. Dalam kelas ini:

      • Nilai vertex merepresentasikan jarak terpendek dari vertex ini ke vertex sumber startVertexId.

      • Metode compute() menggunakan rumus iteratif d[v]=min(d[v], d[u]+weight(u, v)) untuk menghitung jarak terpendek dan memperbarui nilai vertex saat ini.

      • Metode cleanup() menulis jarak terpendek dari vertex saat ini ke vertex sumber ke tabel output.

    • Baris 54: Jika Value dari vertex saat ini (jalur terpendek dari vertex ke vertex sumber) tidak berubah, panggil metode voteToHalt() melalui framework untuk mengubah vertex ke status halt. Ketika semua vertex memasuki status halt, komputasi berakhir.

    • Baris 71: Mendefinisikan GraphLoader untuk memuat data graf sebagai graf berarah. Metode ini mengurai catatan tabel menjadi vertex dan edge graf, lalu memuatnya ke dalam framework. Dalam contoh ini, metode addVertexRequest memuat informasi vertex ke dalam konteks komputasi graf.

    • Baris 90: Mendefinisikan MinLongCombiner. Ini menggabungkan pesan yang dikirim ke vertex yang sama untuk mengoptimalkan kinerja dan mengurangi konsumsi memori.

    • Baris 101: Fungsi main mendefinisikan GraphJob. Fungsi ini mengatur implementasi untuk Vertex, GraphLoader, BaseLoadingVertexResolver, dan Combiner, serta mengonfigurasi tabel input dan output.

    • Baris 110: Mengatur kelas BaseLoadingVertexResolver untuk menangani konflik.

• Graf tak berarah

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

  Penjelasan kode:

  • Baris 15: Mendefinisikan SSSPVertex. Dalam kelas ini:

    • Nilai vertex merepresentasikan jarak terpendek dari vertex ini ke vertex sumber startVertexId.

    • Metode compute() menggunakan rumus iteratif d[v]=min(d[v], d[u]+weight(u, v)) untuk menghitung jarak terpendek dan memperbarui nilai vertex saat ini.

    • Metode cleanup() menulis jarak terpendek dari vertex saat ini ke vertex sumber ke tabel output.

  • Baris 54: Jika Value dari vertex saat ini (jalur terpendek dari vertex ini ke vertex sumber) tidak berubah, vertex memasuki status halt dengan memanggil voteToHalt() melalui framework. Komputasi berakhir ketika semua vertex memasuki status halt.

  • Baris 61: Mendefinisikan MinLongCombiner. Ini menggabungkan pesan yang dikirim ke vertex yang sama untuk mengoptimalkan kinerja dan mengurangi konsumsi memori.

  • Baris 72: Mendefinisikan GraphLoader untuk memuat data graf sebagai graf tak berarah. Metode ini menggunakan addEdgeRequest untuk memuat edge antara dua vertex sebagai edge dua arah, sehingga data tabel dimuat sebagai graf tak berarah.

    • Baris 80: Kolom pertama merepresentasikan ID vertex sumber.

    • Baris 81: Kolom kedua merepresentasikan ID vertex tujuan.

    • Baris 82: Kolom ketiga merepresentasikan bobot edge.

    • Baris 83: Membuat edge yang terdiri dari ID vertex tujuan dan bobot edge.

    • Baris 84: Mengajukan permintaan untuk menambahkan edge ke vertex sumber.

    • Baris 85–87: Setiap Record merepresentasikan edge dua arah. Logika dari baris 83 dan 84 diulang untuk arah sebaliknya.

  • Mendefinisikan SSSPLoadingVertexResolver. Kelas ini menangani konflik yang terjadi saat memuat data untuk graf tak berarah. Misalnya, jika edge yang sama ditambahkan dua kali melalui dua operasi addEdgeRequest, terjadi konflik pemuatan. Anda harus menangani edge duplikat untuk memastikan komputasi yang benar.

  • Baris 101: Fungsi main mendefinisikan GraphJob. Fungsi ini mengatur implementasi untuk Vertex, GraphLoader, SSSPLoadingVertexResolver, dan Combiner, serta mengonfigurasi tabel input dan output.

Hasil eksekusi

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

Tutorial

Untuk informasi lebih lanjut tentang cara mengimplementasikan contoh kode di atas, lihat Mengembangkan program Graph.