Giraph 源码分析（五）—— 加载数据+同步总结

//Znode的路径  
   String addressesAndPartitionsPath =  
       getAddressesAndPartitionsPath(getApplicationAttempt(),  
           getSuperstep());  
//把该znode的data读入到addressesAndPartitions中  
   AddressesAndPartitionsWritable addressesAndPartitions =  
       new AddressesAndPartitionsWritable(  
           workerGraphPartitioner.createPartitionOwner().getClass());  
 //当master创建该znode后，退出while循环  
    while (getZkExt().exists(addressesAndPartitionsPath, true) ==  
         null) {  
     //陷入等待状态  
       getAddressesAndPartitionsReadyChangedEvent().waitForever();  
    //当master创建该znode后，触发Watcher。调用process进而唤醒线程  
       getAddressesAndPartitionsReadyChangedEvent().reset();  
     }  
  //读入数据  
     WritableUtils.readFieldsFromZnode(  
         getZkExt(),  
         addressesAndPartitionsPath,  
         false,  
         null,  
         addressesAndPartitions); 

    List<InputSplit> splits=inputFormat.getSplits(getContext(), minSplitCountHint);  

    @Override  
      public Collection<PartitionOwner> createInitialPartitionOwners(  
          Collection<WorkerInfo> availableWorkerInfos, int maxWorkers) {  
          //maxWorkers为Workers的最大数目，用户通过 -w 指定。实验时指定为 2  
          //availableWorkerInfos为健康的Workers列表,此处为：[Worker(hostname=giraphx, MRtaskID=1, port=30001), Worker(hostname=giraphx, MRtaskID=2, port=30002)]  
        int partitionCount = PartitionUtils.computePartitionCount(  
            availableWorkerInfos, maxWorkers, conf);  
        List<PartitionOwner> ownerList = new ArrayList<PartitionOwner>();  
        Iterator<WorkerInfo> workerIt = availableWorkerInfos.iterator();  
        //为每个Partition指定一个PartitionOwner，表示该Partition的元数据信息  
        for (int i = 0; i < partitionCount; ++i) {  
          PartitionOwner owner = new BasicPartitionOwner(i, workerIt.next());  
          //若遍历完availableWorkerInfos，则开始下一轮遍历。  
          if (!workerIt.hasNext()) {  
            workerIt = availableWorkerInfos.iterator();  
          }  
          ownerList.add(owner);  
        }  
        this.partitionOwnerList = ownerList;  
        return ownerList;  
      }  

public String reserveInputSplit() {  
 String reservedInputSplitPath;  
 Stat reservedStat;  
  
 while (true) {  
//currentIndex递增，要遍历完pathList  
   int splitToTry = currentIndex.getAndIncrement();  
//遍历完pathList，说明所有的split都被处理了，退出while循环。  
   if (splitToTry >= pathList.size()) {  
     return null;  
   }  
//得到split的znode path，如/_hadoopBsp/job_201404102333_0013/_vertexInputSplitDir/1  
   String nextSplitToClaim = pathList.get(splitToTry);  
  
//构造znode路径，如： /_hadoopBsp/job_201404102333_0013/_vertexInputSplitDir/1/_vertexInputSplitReserved  
   String tmpInputSplitReservedPath = nextSplitToClaim + inputSplitReservedNode;  
     
//检测znode是否存在。若存在，说明该split已经被其他worker处理了。设置watcher是为了容错，可暂时忽略。  
   reservedStat =  
       zooKeeper.exists(tmpInputSplitReservedPath, this);  
//若不存在，说明该znode对应的split还没有被处理掉。但有可能其他worker也在申请处理当前znode，  
//所以下面创建znode时,可能会出现KeeperException.NodeExistsException异常。  
   if (reservedStat == null) {  
     try {  
       // Attempt to reserve this InputSplit  
 //若成功创建，那么当前worker就出该split  
       zooKeeper.createExt(tmpInputSplitReservedPath,  
           null,  
           ZooDefs.Ids.OPEN_ACL_UNSAFE,  
           CreateMode.EPHEMERAL,  
           false);  
       reservedInputSplitPath = nextSplitToClaim;  
       return reservedInputSplitPath;  
     } catch (KeeperException.NodeExistsException e) {  
       LOG.info("reserveInputSplit: Couldn't reserve " +  
           "(already reserved) inputSplit" +  
           " at " + tmpInputSplitReservedPath);  
     }  
   }  
 }  

    //inputSplitPath为znode的path，如： /_hadoopBsp/job_201404102333_0013/_vertexInputSplitDir/1  
    private VertexEdgeCount loadInputSplit(  
         String inputSplitPath,  
         GraphState<I, V, E, M> graphState) {  
    //获取该znode对应的InputSplit信息，如得到：hdfs://giraphx:9000/user/root/SSSP/2.txt:0+46  
       InputSplit inputSplit = getInputSplit(inputSplitPath);  
         
    //从split中一行一行读入数据，把每行数据创建成一个vertex。  
    //然后根据vertexId把vertex发送到相应的partition上(数据重分布过程)  
       VertexEdgeCount vertexEdgeCount =  
           readInputSplit(inputSplit, graphState);  
      //处理完当前split后，创建结束znode标识该split已被处理掉。znode的path为： /_hadoopBsp/job_201404102333_0013/_vertexInputSplitDir/1/_vertexInputSplitFinished  
       splitsHandler.markInputSplitPathFinished(inputSplitPath);  
       return vertexEdgeCount;  
     }  

    protected VertexEdgeCount readInputSplit(  
         InputSplit inputSplit,  
         GraphState<I, V, E, M> graphState)  
       throws IOException, InterruptedException {  
    //获取用户输入的InputFormat类  
       VertexInputFormat<I, V, E> vertexInputFormat =  
           configuration.createVertexInputFormat();  
       VertexReader<I, V, E> vertexReader =  
           vertexInputFormat.createVertexReader(inputSplit, context);  
       vertexReader.setConf(  
           (ImmutableClassesGiraphConfiguration<I, V, E, Writable>) configuration);  
       vertexReader.initialize(inputSplit, context);  
       long inputSplitVerticesLoaded = 0;  
       long edgesSinceLastUpdate = 0;  
       long inputSplitEdgesLoaded = 0;  
       while (vertexReader.nextVertex()) {  
      //获取vertex  
         Vertex<I, V, E, M> readerVertex =  
             (Vertex<I, V, E, M>) vertexReader.getCurrentVertex();  
         if (readerVertex.getId() == null) {  
           throw new IllegalArgumentException(  
               "readInputSplit: Vertex reader returned a vertex " +  
                   "without an id!  - " + readerVertex);  
         }  
         if (readerVertex.getValue() == null) {  
           readerVertex.setValue(configuration.createVertexValue());  
         }  
         readerVertex.setConf(configuration);  
         readerVertex.setGraphState(graphState);  
         //根据vertexID获取其partitionOwner  
         PartitionOwner partitionOwner =  
             bspServiceWorker.getVertexPartitionOwner(readerVertex.getId());  
         //把顶点发送到对应的partition上  
         graphState.getWorkerClientRequestProcessor().sendVertexRequest(  
             partitionOwner, readerVertex);  
         context.progress(); // do this before potential data transfer  
         ++inputSplitVerticesLoaded;  
          
         edgesSinceLastUpdate += readerVertex.getNumEdges();  
       }  
       vertexReader.close();  
       return new VertexEdgeCount(inputSplitVerticesLoaded,  
           inputSplitEdgesLoaded + edgesSinceLastUpdate);  
     }