MaxCompute:Tek kaynak en kısa yol

Nasıl çalışır

Dijkstra algoritması, vertices kullanarak shortest distance values günceller. Her vertex, source vertex den kendisine olan mevcut shortest distance value saklar. Bu değer değiştiğinde, düğüm bir edge weight yeni değere ekler ve adjacent vertices bildirimde bulunmak üzere bir mesaj gönderir. Bir sonraki yinelemede, adjacent vertices alınan mesajlara göre mevcut shortest distance values günceller. Tüm vertices mevcut shortest distance values artık değişmediğinde yineleme sona erer.

• Başlatma: Kaynak düğüm s den kendisine olan mesafe 0'dır (d[s]=0) ve diğer herhangi bir düğüm u dan s ye olan mesafe sonsuzdur (d[u]=∞).

• Yineleme: u dan v ye bir kenar varsa, s den v ye olan en kısa mesafe şu şekilde güncellenir: d[v]=min(d[v], d[u]+weight(u, v)). s den diğer tüm düğümlere olan mesafeler kararlı hale gelene kadar işlem devam eder.

Bu algoritma, MaxCompute Graph programlama modeline doğal olarak uyar.

Kullanım senaryoları

MaxCompute hem yönlü hem de yönsüz grafikleri destekler. Yol kullanılabilirliği kaynak verilere ve grafik yapısına bağlı olduğundan, SSSP sonuçları grafik türlerine göre farklılık gösterebilir. Yönlü grafik, tüm hesaplamaların temel aldığı temel veri modelidir.

Kod örnekleri

Aşağıdaki örnekler, yönlü ve yönsüz grafikler için SSSP uygulamalarını göstermektedir.

• Yönlü grafik

  • BaseLoadingVertexResolver sınıfını tanımlayın. Bu sınıfa ana SSSP sınıfından başvurulur.

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

    Kod açıklaması:

    • Satır 15: BaseLoadingVertexResolver tanımlanır. Bu sınıf, yönlü bir grafik için veri yüklenirken oluşan çakışmaları yönetir.

    • Satır 18: resolve yöntemi, bir çakışmayı ele alma mantığını içerir. Örneğin, bir düğüm iki addVertexRequest işlemi aracılığıyla iki kez eklenirse yükleme çakışması oluşur. Hesaplamanın devam edebilmesi için bu çakışmayı çözmeniz gerekir.

  • SSSP sınıfını tanımlayın.

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

    Kod açıklaması:

    • Satır 19: SSSPVertex tanımlanır. Bu sınıfta:

      • Düğüm değeri, bu düğümden kaynak düğüm startVertexId ye olan en kısa mesafeyi temsil eder.

      • compute() yöntemi, en kısa mesafeyi hesaplamak ve mevcut düğümün değerini güncellemek için d[v]=min(d[v], d[u]+weight(u, v)) yineleme formülünü kullanır.

      • cleanup() yöntemi, mevcut düğümden kaynak düğüme olan en kısa mesafeyi çıkış tablosuna yazar.

    • Satır 54: Mevcut düğümün Value değeri (düğümün kaynak düğüme olan en kısa yolu) değişmezse, düğümü halt durumuna geçirmek için framework üzerinden voteToHalt() yöntemini çağırın. Tüm düğümler halt durumuna geçtiğinde hesaplama sona erer.

    • Satır 71: Grafik verilerini yönlü bir grafik olarak yüklemek için bir GraphLoader tanımlar. Tablo kayıtlarını grafik düğümlerine ve kenarlarına ayrıştırır ve bunları framework'e yükler. Bu örnekte, addVertexRequest yöntemi düğüm bilgilerini grafiğin hesaplama bağlamına yükler.

    • Satır 90: MinLongCombiner tanımlanır. Bu sınıf, aynı düğüme gönderilen mesajları birleştirerek performansı optimize eder ve bellek tüketimini azaltır.

    • Satır 101: main işlevi GraphJob tanımını içerir. Vertex, GraphLoader, BaseLoadingVertexResolver ve Combiner uygulamalarını ayarlar ve giriş/çıkış tablolarını yapılandırır.

    • Satır 110: Çakışmaları yönetmek için BaseLoadingVertexResolver sınıfını ayarlar.

• Yönsüz grafik

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

  Kod açıklaması:

  • Satır 15: SSSPVertex tanımlanır. Bu sınıfta:

    • Düğüm değeri, bu düğümden kaynak düğüm startVertexId ye olan en kısa mesafeyi temsil eder.

    • compute() yöntemi, en kısa mesafeyi hesaplamak ve mevcut düğümün değerini güncellemek için d[v]=min(d[v], d[u]+weight(u, v)) yineleme formülünü kullanır.

    • cleanup() yöntemi, mevcut düğümden kaynak düğüme olan en kısa mesafeyi çıkış tablosuna yazar.

  • Satır 54: Mevcut düğümün Value değeri (bu düğümün kaynak düğüme olan en kısa yolu) değişmezse, framework üzerinden voteToHalt() çağrılarak düğüm halt durumuna geçer. Tüm düğümler halt durumuna geçtiğinde hesaplama sona erer.

  • Satır 61: MinLongCombiner tanımlanır. Bu sınıf, aynı düğüme gönderilen mesajları birleştirerek performansı optimize eder ve bellek tüketimini azaltır.

  • Satır 72: Grafik verilerini yönsüz bir grafik olarak yüklemek için bir GraphLoader tanımlar. İki düğüm arasındaki kenarı çift yönlü bir kenar olarak yüklemek için addEdgeRequest kullanır; böylece tablo verileri yönsüz bir grafik olarak yüklenir.

    • Satır 80: İlk sütun kaynak düğüm ID'sini temsil eder.

    • Satır 81: İkinci sütun hedef düğüm ID'sini temsil eder.

    • Satır 82: Üçüncü sütun kenar ağırlığını temsil eder.

    • Satır 83: Hedef düğüm ID'si ve kenar ağırlığından oluşan bir kenar oluşturur.

    • Satır 84: Kenarı kaynak düğüme eklemek için istek gönderir.

    • Satır 85-87: Her Record çift yönlü bir kenarı temsil eder. Satır 83 ve 84'teki mantık ters yön için tekrarlanır.

  • SSSPLoadingVertexResolver tanımlanır. Bu sınıf, yönsüz bir grafik için veri yüklenirken oluşan çakışmaları yönetir. Örneğin, aynı kenar iki addEdgeRequest işlemi aracılığıyla iki kez eklenirse yükleme çakışması oluşur. Doğru hesaplama sağlamak için yinelenen kenarları ele almanız gerekir.

  • Satır 101: main işlevi GraphJob tanımını içerir. Vertex, GraphLoader, SSSPLoadingVertexResolver ve Combiner uygulamalarını ayarlar ve giriş/çıkış tablolarını yapılandırır.

Çalıştırma sonuçları

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

Öğretici

Önceki kod örneklerinin uygulanması hakkında daha fazla bilgi için bkz. Grafik programları geliştirme.