giraph-core/src/main/java/org/apache/giraph/worker/EdgeInputSplitsCallable.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.graph.VertexEdgeCount;
 import org.apache.giraph.io.EdgeInputFormat;
 import org.apache.giraph.io.EdgeReader;
 import org.apache.giraph.io.filters.EdgeInputFilter;
 import org.apache.giraph.io.InputType;
 import org.apache.giraph.ooc.OutOfCoreEngine;
 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 edge input splits as possible.
  * Every thread will has its own instance of WorkerClientRequestProcessor
  * to send requests.
  *
  * @param <I> Vertex id
  * @param <V> Vertex value
  * @param <E> Edge value
  */
 @SuppressWarnings("unchecked")
 public class EdgeInputSplitsCallable<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 EDGES_UPDATE_PERIOD = 1000000;
   /** How often to update filtered metrics */
   public static final int EDGES_FILTERED_UPDATE_PERIOD = 10000;

   /** Class logger */
   private static final Logger LOG = Logger.getLogger(
       EdgeInputSplitsCallable.class);

   /** Aggregator handler */
   private final WorkerThreadGlobalCommUsage globalCommUsage;
   /** Bsp service worker (only use thread-safe methods) */
   private final BspServiceWorker<I, V, E> bspServiceWorker;
   /** Edge input format */
   private final EdgeInputFormat<I, E> edgeInputFormat;
   /** Input split max edges (-1 denotes all) */
   private final long inputSplitMaxEdges;
   /** Can embedInfo in vertexIds */
   private final boolean canEmbedInIds;

   /** Filter to use */
   private final EdgeInputFilter<I, E> edgeInputFilter;

   // Metrics
   /** edges loaded meter across all readers */
   private final Meter totalEdgesMeter;
   /** edges filtered out by user */
   private final Counter totalEdgesFiltered;

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

     this.bspServiceWorker = bspServiceWorker;
     inputSplitMaxEdges = configuration.getInputSplitMaxEdges();
     // Initialize aggregator usage.
     this.globalCommUsage = bspServiceWorker.getAggregatorHandler()
       .newThreadAggregatorUsage();
     edgeInputFilter = configuration.getEdgeInputFilter();
     canEmbedInIds = bspServiceWorker
         .getLocalData()
         .getMappingStoreOps() != null &&
         bspServiceWorker
             .getLocalData()
             .getMappingStoreOps()
             .hasEmbedding();

     // Initialize Metrics
     totalEdgesMeter = getTotalEdgesLoadedMeter();
     totalEdgesFiltered = getTotalEdgesFilteredCounter();
   }

   @Override
   public EdgeInputFormat<I, E> getInputFormat() {
     return edgeInputFormat;
   }

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

   /**
    * Read edges from input split.  If testing, the user may request a
    * maximum number of edges to be read from an input split.
    *
    * @param inputSplit Input split to process with edge reader
    * @return Edges loaded from this input split
    * @throws IOException
    * @throws InterruptedException
    */
   @Override
   protected VertexEdgeCount readInputSplit(
       InputSplit inputSplit) throws IOException,
       InterruptedException {
     EdgeReader<I, E> edgeReader =
         edgeInputFormat.createEdgeReader(inputSplit, context);
     edgeReader.setConf(
         (ImmutableClassesGiraphConfiguration<I, Writable, E>)
             configuration);

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

     long inputSplitEdgesLoaded = 0;
     long inputSplitEdgesFiltered = 0;

     int count = 0;
     OutOfCoreEngine oocEngine = bspServiceWorker.getServerData().getOocEngine();
     while (edgeReader.nextEdge()) {
       // 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();
       }
       I sourceId = edgeReader.getCurrentSourceId();
       Edge<I, E> readerEdge = edgeReader.getCurrentEdge();
       if (sourceId == null) {
         throw new IllegalArgumentException(
             "readInputSplit: Edge reader returned an edge " +
                 "without a source vertex id!  - " + readerEdge);
       }
       if (readerEdge.getTargetVertexId() == null) {
         throw new IllegalArgumentException(
             "readInputSplit: Edge reader returned an edge " +
                 "without a target vertex id!  - " + readerEdge);
       }
       if (readerEdge.getValue() == null) {
         throw new IllegalArgumentException(
             "readInputSplit: Edge reader returned an edge " +
                 "without a value!  - " + readerEdge);
       }
       if (canEmbedInIds) {
         bspServiceWorker
             .getLocalData()
             .getMappingStoreOps()
             .embedTargetInfo(sourceId);
         bspServiceWorker
             .getLocalData()
             .getMappingStoreOps()
             .embedTargetInfo(readerEdge.getTargetVertexId());
       }

       ++inputSplitEdgesLoaded;

       if (edgeInputFilter.dropEdge(sourceId, readerEdge)) {
         ++inputSplitEdgesFiltered;
         if (inputSplitEdgesFiltered % EDGES_FILTERED_UPDATE_PERIOD == 0) {
           totalEdgesFiltered.inc(inputSplitEdgesFiltered);
           inputSplitEdgesFiltered = 0;
         }
         continue;
       }

       workerClientRequestProcessor.sendEdgeRequest(sourceId, readerEdge);

       // Update status every EDGES_UPDATE_PERIOD edges
       if (inputSplitEdgesLoaded % EDGES_UPDATE_PERIOD == 0) {
         totalEdgesMeter.mark(EDGES_UPDATE_PERIOD);
         WorkerProgress.get().addEdgesLoaded(EDGES_UPDATE_PERIOD);
         LoggerUtils.setStatusAndLog(context, LOG, Level.INFO,
             "readEdgeInputSplit: Loaded " +
                 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 (inputSplitMaxEdges > 0 &&
           inputSplitEdgesLoaded >= inputSplitMaxEdges) {
         if (LOG.isInfoEnabled()) {
           LOG.info("readInputSplit: Leaving the input " +
               "split early, reached maximum edges " +
               inputSplitEdgesLoaded);
         }
         break;
       }
     }
     edgeReader.close();

     totalEdgesFiltered.inc(inputSplitEdgesFiltered);
     totalEdgesMeter.mark(inputSplitEdgesLoaded % EDGES_UPDATE_PERIOD);

     WorkerProgress.get().addEdgesLoaded(
         inputSplitEdgesLoaded % EDGES_UPDATE_PERIOD);
     WorkerProgress.get().incrementEdgeInputSplitsLoaded();

     return new VertexEdgeCount(0, inputSplitEdgesLoaded, 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.graph.VertexEdgeCount;
	import org.apache.giraph.io.EdgeInputFormat;
	import org.apache.giraph.io.EdgeReader;
	import org.apache.giraph.io.filters.EdgeInputFilter;
	import org.apache.giraph.io.InputType;
	import org.apache.giraph.ooc.OutOfCoreEngine;
	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 edge input splits as possible.
	* Every thread will has its own instance of WorkerClientRequestProcessor
	* to send requests.
	*
	* @param <I> Vertex id
	* @param <V> Vertex value
	* @param <E> Edge value
	*/
	@SuppressWarnings("unchecked")
	public class EdgeInputSplitsCallable<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 EDGES_UPDATE_PERIOD = 1000000;
	/** How often to update filtered metrics */
	public static final int EDGES_FILTERED_UPDATE_PERIOD = 10000;

	/** Class logger */
	private static final Logger LOG = Logger.getLogger(
	EdgeInputSplitsCallable.class);

	/** Aggregator handler */
	private final WorkerThreadGlobalCommUsage globalCommUsage;
	/** Bsp service worker (only use thread-safe methods) */
	private final BspServiceWorker<I, V, E> bspServiceWorker;
	/** Edge input format */
	private final EdgeInputFormat<I, E> edgeInputFormat;
	/** Input split max edges (-1 denotes all) */
	private final long inputSplitMaxEdges;
	/** Can embedInfo in vertexIds */
	private final boolean canEmbedInIds;

	/** Filter to use */
	private final EdgeInputFilter<I, E> edgeInputFilter;

	// Metrics
	/** edges loaded meter across all readers */
	private final Meter totalEdgesMeter;
	/** edges filtered out by user */
	private final Counter totalEdgesFiltered;

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

	this.bspServiceWorker = bspServiceWorker;
	inputSplitMaxEdges = configuration.getInputSplitMaxEdges();
	// Initialize aggregator usage.
	this.globalCommUsage = bspServiceWorker.getAggregatorHandler()
	.newThreadAggregatorUsage();
	edgeInputFilter = configuration.getEdgeInputFilter();
	canEmbedInIds = bspServiceWorker
	.getLocalData()
	.getMappingStoreOps() != null &&
	bspServiceWorker
	.getLocalData()
	.getMappingStoreOps()
	.hasEmbedding();

	// Initialize Metrics
	totalEdgesMeter = getTotalEdgesLoadedMeter();
	totalEdgesFiltered = getTotalEdgesFilteredCounter();
	}

	@Override
	public EdgeInputFormat<I, E> getInputFormat() {
	return edgeInputFormat;
	}

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

	/**
	* Read edges from input split. If testing, the user may request a
	* maximum number of edges to be read from an input split.
	*
	* @param inputSplit Input split to process with edge reader
	* @return Edges loaded from this input split
	* @throws IOException
	* @throws InterruptedException
	*/
	@Override
	protected VertexEdgeCount readInputSplit(
	InputSplit inputSplit) throws IOException,
	InterruptedException {
	EdgeReader<I, E> edgeReader =
	edgeInputFormat.createEdgeReader(inputSplit, context);
	edgeReader.setConf(
	(ImmutableClassesGiraphConfiguration<I, Writable, E>)
	configuration);

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

	long inputSplitEdgesLoaded = 0;
	long inputSplitEdgesFiltered = 0;

	int count = 0;
	OutOfCoreEngine oocEngine = bspServiceWorker.getServerData().getOocEngine();
	while (edgeReader.nextEdge()) {
	// 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();
	}
	I sourceId = edgeReader.getCurrentSourceId();
	Edge<I, E> readerEdge = edgeReader.getCurrentEdge();
	if (sourceId == null) {
	throw new IllegalArgumentException(
	"readInputSplit: Edge reader returned an edge " +
	"without a source vertex id! - " + readerEdge);
	}
	if (readerEdge.getTargetVertexId() == null) {
	throw new IllegalArgumentException(
	"readInputSplit: Edge reader returned an edge " +
	"without a target vertex id! - " + readerEdge);
	}
	if (readerEdge.getValue() == null) {
	throw new IllegalArgumentException(
	"readInputSplit: Edge reader returned an edge " +
	"without a value! - " + readerEdge);
	}
	if (canEmbedInIds) {
	bspServiceWorker
	.getLocalData()
	.getMappingStoreOps()
	.embedTargetInfo(sourceId);
	bspServiceWorker
	.getLocalData()
	.getMappingStoreOps()
	.embedTargetInfo(readerEdge.getTargetVertexId());
	}

	++inputSplitEdgesLoaded;

	if (edgeInputFilter.dropEdge(sourceId, readerEdge)) {
	++inputSplitEdgesFiltered;
	if (inputSplitEdgesFiltered % EDGES_FILTERED_UPDATE_PERIOD == 0) {
	totalEdgesFiltered.inc(inputSplitEdgesFiltered);
	inputSplitEdgesFiltered = 0;
	}
	continue;
	}

	workerClientRequestProcessor.sendEdgeRequest(sourceId, readerEdge);

	// Update status every EDGES_UPDATE_PERIOD edges
	if (inputSplitEdgesLoaded % EDGES_UPDATE_PERIOD == 0) {
	totalEdgesMeter.mark(EDGES_UPDATE_PERIOD);
	WorkerProgress.get().addEdgesLoaded(EDGES_UPDATE_PERIOD);
	LoggerUtils.setStatusAndLog(context, LOG, Level.INFO,
	"readEdgeInputSplit: Loaded " +
	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 (inputSplitMaxEdges > 0 &&
	inputSplitEdgesLoaded >= inputSplitMaxEdges) {
	if (LOG.isInfoEnabled()) {
	LOG.info("readInputSplit: Leaving the input " +
	"split early, reached maximum edges " +
	inputSplitEdgesLoaded);
	}
	break;
	}
	}
	edgeReader.close();

	totalEdgesFiltered.inc(inputSplitEdgesFiltered);
	totalEdgesMeter.mark(inputSplitEdgesLoaded % EDGES_UPDATE_PERIOD);

	WorkerProgress.get().addEdgesLoaded(
	inputSplitEdgesLoaded % EDGES_UPDATE_PERIOD);
	WorkerProgress.get().incrementEdgeInputSplitsLoaded();

	return new VertexEdgeCount(0, inputSplitEdgesLoaded, 0);
	}
	}