giraph-core/src/main/java/org/apache/giraph/worker/VertexInputSplitsCallable.java - giraph - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.giraph.worker;

 import java.io.IOException;

 import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
 import org.apache.giraph.edge.Edge;
 import org.apache.giraph.edge.OutEdges;
 import org.apache.giraph.graph.Vertex;
 import org.apache.giraph.graph.VertexEdgeCount;
 import org.apache.giraph.io.GiraphInputFormat;
 import org.apache.giraph.io.VertexInputFormat;
 import org.apache.giraph.io.VertexReader;
 import org.apache.giraph.io.filters.VertexInputFilter;
 import org.apache.giraph.mapping.translate.TranslateEdge;
 import org.apache.giraph.io.InputType;
 import org.apache.giraph.ooc.OutOfCoreEngine;
 import org.apache.giraph.partition.PartitionOwner;
 import org.apache.giraph.utils.LoggerUtils;
 import org.apache.giraph.utils.MemoryUtils;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.Mapper;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;

 import com.yammer.metrics.core.Counter;
 import com.yammer.metrics.core.Meter;

 /**
  * Load as many vertex input splits as possible.
  * Every thread will has its own instance of WorkerClientRequestProcessor
  * to send requests.
  *
  * @param <I> Vertex index value
  * @param <V> Vertex value
  * @param <E> Edge value
  */
 @SuppressWarnings("unchecked")
 public class VertexInputSplitsCallable<I extends WritableComparable,
     V extends Writable, E extends Writable>
     extends InputSplitsCallable<I, V, E> {
   /** How often to update metrics and print info */
   public static final int VERTICES_UPDATE_PERIOD = 250000;
   /** How often to update filtered out metrics */
   public static final int VERTICES_FILTERED_UPDATE_PERIOD = 2500;

   /** Class logger */
   private static final Logger LOG =
       Logger.getLogger(VertexInputSplitsCallable.class);
   /** Vertex input format */
   private final VertexInputFormat<I, V, E> vertexInputFormat;
   /** Input split max vertices (-1 denotes all) */
   private final long inputSplitMaxVertices;
   /** Bsp service worker (only use thread-safe methods) */
   private final BspServiceWorker<I, V, E> bspServiceWorker;
   /** Filter to select which vertices to keep */
   private final VertexInputFilter<I, V, E> vertexInputFilter;
   /** Can embedInfo in vertexIds */
   private final boolean canEmbedInIds;
   /**
    * Whether the chosen {@link OutEdges} implementation allows for Edge
    * reuse.
    */
   private final boolean reuseEdgeObjects;
   /** Used to translate Edges during vertex input phase based on localData */
   private final TranslateEdge<I, E> translateEdge;

   // Metrics
   /** number of vertices loaded meter across all readers */
   private final Meter totalVerticesMeter;
   /** number of vertices filtered out */
   private final Counter totalVerticesFilteredCounter;
   /** number of edges loaded meter across all readers */
   private final Meter totalEdgesMeter;

   /**
    * Constructor.
    *
    * @param vertexInputFormat Vertex input format
    * @param context Context
    * @param configuration Configuration
    * @param bspServiceWorker service worker
    * @param splitsHandler Handler for input splits
    */
   public VertexInputSplitsCallable(
       VertexInputFormat<I, V, E> vertexInputFormat,
       Mapper<?, ?, ?, ?>.Context context,
       ImmutableClassesGiraphConfiguration<I, V, E> configuration,
       BspServiceWorker<I, V, E> bspServiceWorker,
       WorkerInputSplitsHandler splitsHandler)  {
     super(context, configuration, bspServiceWorker, splitsHandler);
     this.vertexInputFormat = vertexInputFormat;

     inputSplitMaxVertices = configuration.getInputSplitMaxVertices();
     this.bspServiceWorker = bspServiceWorker;
     vertexInputFilter = configuration.getVertexInputFilter();
     reuseEdgeObjects = configuration.reuseEdgeObjects();
     canEmbedInIds = bspServiceWorker
         .getLocalData()
         .getMappingStoreOps() != null &&
         bspServiceWorker
             .getLocalData()
             .getMappingStoreOps()
             .hasEmbedding();
     translateEdge = bspServiceWorker.getTranslateEdge();

     // Initialize Metrics
     totalVerticesMeter = getTotalVerticesLoadedMeter();
     totalVerticesFilteredCounter = getTotalVerticesFilteredCounter();
     totalEdgesMeter = getTotalEdgesLoadedMeter();
   }

   @Override
   public GiraphInputFormat getInputFormat() {
     return vertexInputFormat;
   }

   @Override
   public InputType getInputType() {
     return InputType.VERTEX;
   }

   /**
    * Read vertices from input split.  If testing, the user may request a
    * maximum number of vertices to be read from an input split.
    *
    * @param inputSplit Input split to process with vertex reader
    * @return Vertices and edges loaded from this input split
    * @throws IOException
    * @throws InterruptedException
    */
   @Override
   protected VertexEdgeCount readInputSplit(
       InputSplit inputSplit) throws IOException, InterruptedException {
     VertexReader<I, V, E> vertexReader =
         vertexInputFormat.createVertexReader(inputSplit, context);
     vertexReader.setConf(configuration);

     WorkerThreadGlobalCommUsage globalCommUsage =
       this.bspServiceWorker
         .getAggregatorHandler().newThreadAggregatorUsage();

     vertexReader.initialize(inputSplit, context);
     // Set aggregator usage to vertex reader
     vertexReader.setWorkerGlobalCommUsage(globalCommUsage);

     long inputSplitVerticesLoaded = 0;
     long inputSplitVerticesFiltered = 0;

     long edgesSinceLastUpdate = 0;
     long inputSplitEdgesLoaded = 0;

     int count = 0;
     OutOfCoreEngine oocEngine = bspServiceWorker.getServerData().getOocEngine();
     while (vertexReader.nextVertex()) {
       // If out-of-core mechanism is used, check whether this thread
       // can stay active or it should temporarily suspend and stop
       // processing and generating more data for the moment.
       if (oocEngine != null &&
           (++count & OutOfCoreEngine.CHECK_IN_INTERVAL) == 0) {
         oocEngine.activeThreadCheckIn();
       }
       Vertex<I, V, E> readerVertex = vertexReader.getCurrentVertex();
       if (readerVertex.getId() == null) {
         throw new IllegalArgumentException(
             "readInputSplit: Vertex reader returned a vertex " +
                 "without an id!  - " + readerVertex);
       }
       if (canEmbedInIds) {
         bspServiceWorker
             .getLocalData()
             .getMappingStoreOps()
             .embedTargetInfo(readerVertex.getId());
       }
       if (readerVertex.getValue() == null) {
         readerVertex.setValue(configuration.createVertexValue());
       }
       readerVertex.setConf(configuration);

       ++inputSplitVerticesLoaded;

       if (vertexInputFilter.dropVertex(readerVertex)) {
         ++inputSplitVerticesFiltered;
         if (inputSplitVerticesFiltered % VERTICES_FILTERED_UPDATE_PERIOD == 0) {
           totalVerticesFilteredCounter.inc(inputSplitVerticesFiltered);
           inputSplitVerticesFiltered = 0;
         }
         continue;
       }

       // Before saving to partition-store translate all edges (if present)
       if (translateEdge != null) {
         // only iff vertexInput reads edges also
         if (readerVertex.getEdges() != null && readerVertex.getNumEdges() > 0) {
           OutEdges<I, E> vertexOutEdges = configuration
               .createAndInitializeOutEdges(readerVertex.getNumEdges());
           // TODO : this works for generic OutEdges, can create a better api
           // to support more efficient translation for specific types

           // NOTE : for implementations where edge is reusable, space is
           // consumed by the OutEdges data structure itself, but if not reusable
           // space is consumed by the newly created edge -> and the new OutEdges
           // data structure just holds a reference to the newly created edge
           // so in any way we virtually hold edges twice - similar to
           // OutEdges.trim() -> this has the same complexity as OutEdges.trim()
           for (Edge<I, E> edge : readerVertex.getEdges()) {
             if (reuseEdgeObjects) {
               bspServiceWorker
                   .getLocalData()
                   .getMappingStoreOps()
                   .embedTargetInfo(edge.getTargetVertexId());
               vertexOutEdges.add(edge); // edge can be re-used
             } else { // edge objects cannot be reused - so create new edges
               vertexOutEdges.add(configuration.createEdge(translateEdge, edge));
             }
           }
           // set out edges to translated instance -> old instance is released
           readerVertex.setEdges(vertexOutEdges);
         }
       }

       PartitionOwner partitionOwner =
           bspServiceWorker.getVertexPartitionOwner(readerVertex.getId());
       workerClientRequestProcessor.sendVertexRequest(
           partitionOwner, readerVertex);
       edgesSinceLastUpdate += readerVertex.getNumEdges();

       // Update status every VERTICES_UPDATE_PERIOD vertices
       if (inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD == 0) {
         totalVerticesMeter.mark(VERTICES_UPDATE_PERIOD);
         WorkerProgress.get().addVerticesLoaded(VERTICES_UPDATE_PERIOD);
         totalEdgesMeter.mark(edgesSinceLastUpdate);
         inputSplitEdgesLoaded += edgesSinceLastUpdate;
         edgesSinceLastUpdate = 0;

         LoggerUtils.setStatusAndLog(
             context, LOG, Level.INFO,
             "readVertexInputSplit: Loaded " +
                 totalVerticesMeter.count() + " vertices at " +
                 totalVerticesMeter.meanRate() + " vertices/sec " +
                 totalEdgesMeter.count() + " edges at " +
                 totalEdgesMeter.meanRate() + " edges/sec " +
                 MemoryUtils.getRuntimeMemoryStats());
       }

       // For sampling, or to limit outlier input splits, the number of
       // records per input split can be limited
       if (inputSplitMaxVertices > 0 &&
           inputSplitVerticesLoaded >= inputSplitMaxVertices) {
         if (LOG.isInfoEnabled()) {
           LOG.info("readInputSplit: Leaving the input " +
               "split early, reached maximum vertices " +
               inputSplitVerticesLoaded);
         }
         break;
       }
     }

     totalVerticesMeter.mark(inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD);
     totalEdgesMeter.mark(edgesSinceLastUpdate);
     totalVerticesFilteredCounter.inc(inputSplitVerticesFiltered);

     vertexReader.close();

     WorkerProgress.get().addVerticesLoaded(
         inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD);
     WorkerProgress.get().incrementVertexInputSplitsLoaded();

     return new VertexEdgeCount(inputSplitVerticesLoaded,
         inputSplitEdgesLoaded + edgesSinceLastUpdate, 0);
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.giraph.worker;

	import java.io.IOException;

	import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;
	import org.apache.giraph.edge.Edge;
	import org.apache.giraph.edge.OutEdges;
	import org.apache.giraph.graph.Vertex;
	import org.apache.giraph.graph.VertexEdgeCount;
	import org.apache.giraph.io.GiraphInputFormat;
	import org.apache.giraph.io.VertexInputFormat;
	import org.apache.giraph.io.VertexReader;
	import org.apache.giraph.io.filters.VertexInputFilter;
	import org.apache.giraph.mapping.translate.TranslateEdge;
	import org.apache.giraph.io.InputType;
	import org.apache.giraph.ooc.OutOfCoreEngine;
	import org.apache.giraph.partition.PartitionOwner;
	import org.apache.giraph.utils.LoggerUtils;
	import org.apache.giraph.utils.MemoryUtils;
	import org.apache.hadoop.io.Writable;
	import org.apache.hadoop.io.WritableComparable;
	import org.apache.hadoop.mapreduce.InputSplit;
	import org.apache.hadoop.mapreduce.Mapper;
	import org.apache.log4j.Level;
	import org.apache.log4j.Logger;

	import com.yammer.metrics.core.Counter;
	import com.yammer.metrics.core.Meter;

	/**
	* Load as many vertex input splits as possible.
	* Every thread will has its own instance of WorkerClientRequestProcessor
	* to send requests.
	*
	* @param <I> Vertex index value
	* @param <V> Vertex value
	* @param <E> Edge value
	*/
	@SuppressWarnings("unchecked")
	public class VertexInputSplitsCallable<I extends WritableComparable,
	V extends Writable, E extends Writable>
	extends InputSplitsCallable<I, V, E> {
	/** How often to update metrics and print info */
	public static final int VERTICES_UPDATE_PERIOD = 250000;
	/** How often to update filtered out metrics */
	public static final int VERTICES_FILTERED_UPDATE_PERIOD = 2500;

	/** Class logger */
	private static final Logger LOG =
	Logger.getLogger(VertexInputSplitsCallable.class);
	/** Vertex input format */
	private final VertexInputFormat<I, V, E> vertexInputFormat;
	/** Input split max vertices (-1 denotes all) */
	private final long inputSplitMaxVertices;
	/** Bsp service worker (only use thread-safe methods) */
	private final BspServiceWorker<I, V, E> bspServiceWorker;
	/** Filter to select which vertices to keep */
	private final VertexInputFilter<I, V, E> vertexInputFilter;
	/** Can embedInfo in vertexIds */
	private final boolean canEmbedInIds;
	/**
	* Whether the chosen {@link OutEdges} implementation allows for Edge
	* reuse.
	*/
	private final boolean reuseEdgeObjects;
	/** Used to translate Edges during vertex input phase based on localData */
	private final TranslateEdge<I, E> translateEdge;

	// Metrics
	/** number of vertices loaded meter across all readers */
	private final Meter totalVerticesMeter;
	/** number of vertices filtered out */
	private final Counter totalVerticesFilteredCounter;
	/** number of edges loaded meter across all readers */
	private final Meter totalEdgesMeter;

	/**
	* Constructor.
	*
	* @param vertexInputFormat Vertex input format
	* @param context Context
	* @param configuration Configuration
	* @param bspServiceWorker service worker
	* @param splitsHandler Handler for input splits
	*/
	public VertexInputSplitsCallable(
	VertexInputFormat<I, V, E> vertexInputFormat,
	Mapper<?, ?, ?, ?>.Context context,
	ImmutableClassesGiraphConfiguration<I, V, E> configuration,
	BspServiceWorker<I, V, E> bspServiceWorker,
	WorkerInputSplitsHandler splitsHandler) {
	super(context, configuration, bspServiceWorker, splitsHandler);
	this.vertexInputFormat = vertexInputFormat;

	inputSplitMaxVertices = configuration.getInputSplitMaxVertices();
	this.bspServiceWorker = bspServiceWorker;
	vertexInputFilter = configuration.getVertexInputFilter();
	reuseEdgeObjects = configuration.reuseEdgeObjects();
	canEmbedInIds = bspServiceWorker
	.getLocalData()
	.getMappingStoreOps() != null &&
	bspServiceWorker
	.getLocalData()
	.getMappingStoreOps()
	.hasEmbedding();
	translateEdge = bspServiceWorker.getTranslateEdge();

	// Initialize Metrics
	totalVerticesMeter = getTotalVerticesLoadedMeter();
	totalVerticesFilteredCounter = getTotalVerticesFilteredCounter();
	totalEdgesMeter = getTotalEdgesLoadedMeter();
	}

	@Override
	public GiraphInputFormat getInputFormat() {
	return vertexInputFormat;
	}

	@Override
	public InputType getInputType() {
	return InputType.VERTEX;
	}

	/**
	* Read vertices from input split. If testing, the user may request a
	* maximum number of vertices to be read from an input split.
	*
	* @param inputSplit Input split to process with vertex reader
	* @return Vertices and edges loaded from this input split
	* @throws IOException
	* @throws InterruptedException
	*/
	@Override
	protected VertexEdgeCount readInputSplit(
	InputSplit inputSplit) throws IOException, InterruptedException {
	VertexReader<I, V, E> vertexReader =
	vertexInputFormat.createVertexReader(inputSplit, context);
	vertexReader.setConf(configuration);

	WorkerThreadGlobalCommUsage globalCommUsage =
	this.bspServiceWorker
	.getAggregatorHandler().newThreadAggregatorUsage();

	vertexReader.initialize(inputSplit, context);
	// Set aggregator usage to vertex reader
	vertexReader.setWorkerGlobalCommUsage(globalCommUsage);

	long inputSplitVerticesLoaded = 0;
	long inputSplitVerticesFiltered = 0;

	long edgesSinceLastUpdate = 0;
	long inputSplitEdgesLoaded = 0;

	int count = 0;
	OutOfCoreEngine oocEngine = bspServiceWorker.getServerData().getOocEngine();
	while (vertexReader.nextVertex()) {
	// If out-of-core mechanism is used, check whether this thread
	// can stay active or it should temporarily suspend and stop
	// processing and generating more data for the moment.
	if (oocEngine != null &&
	(++count & OutOfCoreEngine.CHECK_IN_INTERVAL) == 0) {
	oocEngine.activeThreadCheckIn();
	}
	Vertex<I, V, E> readerVertex = vertexReader.getCurrentVertex();
	if (readerVertex.getId() == null) {
	throw new IllegalArgumentException(
	"readInputSplit: Vertex reader returned a vertex " +
	"without an id! - " + readerVertex);
	}
	if (canEmbedInIds) {
	bspServiceWorker
	.getLocalData()
	.getMappingStoreOps()
	.embedTargetInfo(readerVertex.getId());
	}
	if (readerVertex.getValue() == null) {
	readerVertex.setValue(configuration.createVertexValue());
	}
	readerVertex.setConf(configuration);

	++inputSplitVerticesLoaded;

	if (vertexInputFilter.dropVertex(readerVertex)) {
	++inputSplitVerticesFiltered;
	if (inputSplitVerticesFiltered % VERTICES_FILTERED_UPDATE_PERIOD == 0) {
	totalVerticesFilteredCounter.inc(inputSplitVerticesFiltered);
	inputSplitVerticesFiltered = 0;
	}
	continue;
	}

	// Before saving to partition-store translate all edges (if present)
	if (translateEdge != null) {
	// only iff vertexInput reads edges also
	if (readerVertex.getEdges() != null && readerVertex.getNumEdges() > 0) {
	OutEdges<I, E> vertexOutEdges = configuration
	.createAndInitializeOutEdges(readerVertex.getNumEdges());
	// TODO : this works for generic OutEdges, can create a better api
	// to support more efficient translation for specific types

	// NOTE : for implementations where edge is reusable, space is
	// consumed by the OutEdges data structure itself, but if not reusable
	// space is consumed by the newly created edge -> and the new OutEdges
	// data structure just holds a reference to the newly created edge
	// so in any way we virtually hold edges twice - similar to
	// OutEdges.trim() -> this has the same complexity as OutEdges.trim()
	for (Edge<I, E> edge : readerVertex.getEdges()) {
	if (reuseEdgeObjects) {
	bspServiceWorker
	.getLocalData()
	.getMappingStoreOps()
	.embedTargetInfo(edge.getTargetVertexId());
	vertexOutEdges.add(edge); // edge can be re-used
	} else { // edge objects cannot be reused - so create new edges
	vertexOutEdges.add(configuration.createEdge(translateEdge, edge));
	}
	}
	// set out edges to translated instance -> old instance is released
	readerVertex.setEdges(vertexOutEdges);
	}
	}

	PartitionOwner partitionOwner =
	bspServiceWorker.getVertexPartitionOwner(readerVertex.getId());
	workerClientRequestProcessor.sendVertexRequest(
	partitionOwner, readerVertex);
	edgesSinceLastUpdate += readerVertex.getNumEdges();

	// Update status every VERTICES_UPDATE_PERIOD vertices
	if (inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD == 0) {
	totalVerticesMeter.mark(VERTICES_UPDATE_PERIOD);
	WorkerProgress.get().addVerticesLoaded(VERTICES_UPDATE_PERIOD);
	totalEdgesMeter.mark(edgesSinceLastUpdate);
	inputSplitEdgesLoaded += edgesSinceLastUpdate;
	edgesSinceLastUpdate = 0;

	LoggerUtils.setStatusAndLog(
	context, LOG, Level.INFO,
	"readVertexInputSplit: Loaded " +
	totalVerticesMeter.count() + " vertices at " +
	totalVerticesMeter.meanRate() + " vertices/sec " +
	totalEdgesMeter.count() + " edges at " +
	totalEdgesMeter.meanRate() + " edges/sec " +
	MemoryUtils.getRuntimeMemoryStats());
	}

	// For sampling, or to limit outlier input splits, the number of
	// records per input split can be limited
	if (inputSplitMaxVertices > 0 &&
	inputSplitVerticesLoaded >= inputSplitMaxVertices) {
	if (LOG.isInfoEnabled()) {
	LOG.info("readInputSplit: Leaving the input " +
	"split early, reached maximum vertices " +
	inputSplitVerticesLoaded);
	}
	break;
	}
	}

	totalVerticesMeter.mark(inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD);
	totalEdgesMeter.mark(edgesSinceLastUpdate);
	totalVerticesFilteredCounter.inc(inputSplitVerticesFiltered);

	vertexReader.close();

	WorkerProgress.get().addVerticesLoaded(
	inputSplitVerticesLoaded % VERTICES_UPDATE_PERIOD);
	WorkerProgress.get().incrementVertexInputSplitsLoaded();

	return new VertexEdgeCount(inputSplitVerticesLoaded,
	inputSplitEdgesLoaded + edgesSinceLastUpdate, 0);
	}
	}