| /* |
| * Copyright 2009-2013 by The Regents of the University of California |
| * Licensed 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 from |
| * |
| * 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 edu.uci.ics.pregelix.api.util; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.List; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FSDataInputStream; |
| import org.apache.hadoop.fs.FSDataOutputStream; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.io.IntWritable; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.hadoop.io.WritableComparable; |
| import org.apache.hadoop.mapreduce.Counters; |
| |
| import edu.uci.ics.hyracks.api.exceptions.HyracksDataException; |
| import edu.uci.ics.pregelix.api.graph.GlobalAggregator; |
| import edu.uci.ics.pregelix.api.graph.MessageCombiner; |
| import edu.uci.ics.pregelix.api.graph.MsgList; |
| import edu.uci.ics.pregelix.api.graph.NormalizedKeyComputer; |
| import edu.uci.ics.pregelix.api.graph.Vertex; |
| import edu.uci.ics.pregelix.api.graph.VertexPartitioner; |
| import edu.uci.ics.pregelix.api.io.VertexInputFormat; |
| import edu.uci.ics.pregelix.api.io.VertexOutputFormat; |
| import edu.uci.ics.pregelix.api.io.WritableSizable; |
| import edu.uci.ics.pregelix.api.job.ICheckpointHook; |
| import edu.uci.ics.pregelix.api.job.IIterationCompleteReporterHook; |
| import edu.uci.ics.pregelix.api.job.PregelixJob; |
| |
| /** |
| * Help to use the configuration to get the appropriate classes or instantiate |
| * them. |
| */ |
| public class BspUtils { |
| |
| public static final String TMP_DIR = "/tmp/"; |
| private static final String COUNTERS_VALUE_ON_ITERATION = ".counters.valueOnIter."; |
| private static final String COUNTERS_LAST_ITERATION_COMPLETED = ".counters.lastIterCompleted"; |
| |
| /** |
| * Get the user's subclassed {@link VertexInputFormat}. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex input format class |
| */ |
| @SuppressWarnings({ "rawtypes", "unchecked" }) |
| public static <I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable> Class<? extends VertexInputFormat<I, V, E, M>> getVertexInputFormatClass( |
| Configuration conf) { |
| return (Class<? extends VertexInputFormat<I, V, E, M>>) conf.getClass(PregelixJob.VERTEX_INPUT_FORMAT_CLASS, |
| null, VertexInputFormat.class); |
| } |
| |
| /** |
| * Create a user vertex input format class |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user vertex input format class |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable> VertexInputFormat<I, V, E, M> createVertexInputFormat( |
| Configuration conf) { |
| Class<? extends VertexInputFormat<I, V, E, M>> vertexInputFormatClass = getVertexInputFormatClass(conf); |
| try { |
| VertexInputFormat<I, V, E, M> inputFormat = vertexInputFormatClass.newInstance(); |
| return inputFormat; |
| } catch (Exception e) { |
| throw new IllegalStateException(e); |
| } |
| } |
| |
| /** |
| * Get the user's subclassed {@link VertexOutputFormat}. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex output format class |
| */ |
| @SuppressWarnings({ "rawtypes", "unchecked" }) |
| public static <I extends WritableComparable, V extends Writable, E extends Writable> Class<? extends VertexOutputFormat<I, V, E>> getVertexOutputFormatClass( |
| Configuration conf) { |
| return (Class<? extends VertexOutputFormat<I, V, E>>) conf.getClass(PregelixJob.VERTEX_OUTPUT_FORMAT_CLASS, |
| null, VertexOutputFormat.class); |
| } |
| |
| /** |
| * Create a user vertex output format class |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user vertex output format class |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <I extends WritableComparable, V extends Writable, E extends Writable> VertexOutputFormat<I, V, E> createVertexOutputFormat( |
| Configuration conf) { |
| Class<? extends VertexOutputFormat<I, V, E>> vertexOutputFormatClass = getVertexOutputFormatClass(conf); |
| try { |
| return vertexOutputFormatClass.newInstance(); |
| } catch (Exception e) { |
| throw new IllegalStateException(e); |
| } |
| } |
| |
| /** |
| * Get the user's subclassed {@link MessageCombiner}. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex combiner class |
| */ |
| @SuppressWarnings({ "rawtypes", "unchecked" }) |
| public static <I extends WritableComparable, M extends WritableSizable, P extends Writable> Class<? extends MessageCombiner<I, M, P>> getMessageCombinerClass( |
| Configuration conf) { |
| return (Class<? extends MessageCombiner<I, M, P>>) conf.getClass(PregelixJob.Message_COMBINER_CLASS, |
| DefaultMessageCombiner.class, MessageCombiner.class); |
| } |
| |
| /** |
| * Get the user's subclassed {@link GlobalAggregator}. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex combiner class |
| */ |
| @SuppressWarnings({ "rawtypes", "unchecked" }) |
| public static <I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable, P extends Writable, F extends Writable> List<Class<? extends GlobalAggregator<I, V, E, M, P, F>>> getGlobalAggregatorClasses( |
| Configuration conf) { |
| String aggStrs = conf.get(PregelixJob.GLOBAL_AGGREGATOR_CLASS); |
| String[] classnames; |
| if (aggStrs == null) { |
| classnames = new String[0]; |
| } else { |
| classnames = aggStrs.split(PregelixJob.COMMA_STR); |
| } |
| try { |
| List<Class<? extends GlobalAggregator<I, V, E, M, P, F>>> classes = new ArrayList<Class<? extends GlobalAggregator<I, V, E, M, P, F>>>(); |
| for (String defaultClass : PregelixJob.DEFAULT_GLOBAL_AGGREGATOR_CLASSES) { |
| classes.add((Class<? extends GlobalAggregator<I, V, E, M, P, F>>) conf.getClassByName(defaultClass)); |
| } |
| for (int i = 0; i < classnames.length; i++) { |
| classes.add((Class<? extends GlobalAggregator<I, V, E, M, P, F>>) conf.getClassByName(classnames[i])); |
| } |
| return classes; |
| } catch (ClassNotFoundException e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| public static String getJobId(Configuration conf) { |
| return conf.get(PregelixJob.JOB_ID); |
| } |
| |
| /** |
| * Create a user vertex combiner class |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user vertex combiner class |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <I extends WritableComparable, M extends WritableSizable, P extends Writable> MessageCombiner<I, M, P> createMessageCombiner( |
| Configuration conf) { |
| Class<? extends MessageCombiner<I, M, P>> vertexCombinerClass = getMessageCombinerClass(conf); |
| try { |
| return vertexCombinerClass.newInstance(); |
| } catch (Exception e) { |
| throw new IllegalStateException(e); |
| } |
| } |
| |
| /** |
| * Create a user-defined normalized key computer class |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user-defined normalized key computer |
| */ |
| public static NormalizedKeyComputer createNormalizedKeyComputer(Configuration conf) { |
| Class<? extends NormalizedKeyComputer> nmkClass = getNormalizedKeyComputerClass(conf); |
| try { |
| return nmkClass.newInstance(); |
| } catch (Exception e) { |
| throw new IllegalStateException(e); |
| } |
| } |
| |
| /** |
| * Create a global aggregator object |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user vertex combiner class |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable, P extends Writable, F extends Writable> List<GlobalAggregator> createGlobalAggregators( |
| Configuration conf) { |
| List<Class<? extends GlobalAggregator<I, V, E, M, P, F>>> globalAggregatorClasses = getGlobalAggregatorClasses(conf); |
| List<GlobalAggregator> aggs = new ArrayList<GlobalAggregator>(); |
| try { |
| for (Class<? extends GlobalAggregator<I, V, E, M, P, F>> globalAggClass : globalAggregatorClasses) { |
| aggs.add(globalAggClass.newInstance()); |
| } |
| return aggs; |
| } catch (Exception e) { |
| throw new IllegalStateException(e); |
| } |
| } |
| |
| /** |
| * Get global aggregator class names |
| * |
| * @param conf |
| * Configuration to check |
| * @return An array of Global aggregator names |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable, P extends Writable, F extends Writable> String[] getGlobalAggregatorClassNames( |
| Configuration conf) { |
| List<Class<? extends GlobalAggregator<I, V, E, M, P, F>>> globalAggregatorClasses = getGlobalAggregatorClasses(conf); |
| String[] aggClassNames = new String[globalAggregatorClasses.size()]; |
| int i = 0; |
| for (Class<? extends GlobalAggregator<I, V, E, M, P, F>> globalAggClass : globalAggregatorClasses) { |
| aggClassNames[i++] = globalAggClass.getName(); |
| } |
| return aggClassNames; |
| } |
| |
| /** |
| * Get global aggregator class names |
| * |
| * @param conf |
| * Configuration to check |
| * @return An array of Global aggregator names |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable, P extends Writable, F extends Writable> String[] getPartialAggregateValueClassNames( |
| Configuration conf) { |
| String[] gloablAggClassNames = getGlobalAggregatorClassNames(conf); |
| String[] partialAggValueClassNames = new String[gloablAggClassNames.length]; |
| int i = 0; |
| for (String globalAggClassName : gloablAggClassNames) { |
| partialAggValueClassNames[i++] = getPartialAggregateValueClass(conf, globalAggClassName).getName(); |
| } |
| return partialAggValueClassNames; |
| } |
| |
| /** |
| * Get the user's subclassed Vertex. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex class |
| */ |
| @SuppressWarnings({ "rawtypes", "unchecked" }) |
| public static <I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable> Class<? extends Vertex<I, V, E, M>> getVertexClass( |
| Configuration conf) { |
| return (Class<? extends Vertex<I, V, E, M>>) conf.getClass(PregelixJob.VERTEX_CLASS, null, Vertex.class); |
| } |
| |
| /** |
| * Create a user vertex |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user vertex |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <I extends WritableComparable, V extends Writable, E extends Writable, M extends WritableSizable> Vertex<I, V, E, M> createVertex( |
| Configuration conf) { |
| Class<? extends Vertex<I, V, E, M>> vertexClass = getVertexClass(conf); |
| try { |
| Vertex<I, V, E, M> vertex = vertexClass.newInstance(); |
| return vertex; |
| } catch (Exception e) { |
| throw new IllegalStateException(e); |
| } |
| |
| } |
| |
| /** |
| * Get the user's subclassed vertex index class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex index class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <I extends Writable> Class<I> getVertexIndexClass(Configuration conf) { |
| return (Class<I>) conf.getClass(PregelixJob.VERTEX_INDEX_CLASS, WritableComparable.class); |
| } |
| |
| /** |
| * Create a user vertex index |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user vertex index |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <I extends WritableComparable> I createVertexIndex(Configuration conf) { |
| Class<I> vertexClass = getVertexIndexClass(conf); |
| try { |
| return vertexClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createVertexIndex: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createVertexIndex: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Get the user's subclassed vertex value class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex value class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <V extends Writable> Class<V> getVertexValueClass(Configuration conf) { |
| return (Class<V>) conf.getClass(PregelixJob.VERTEX_VALUE_CLASS, Writable.class); |
| } |
| |
| /** |
| * Create a user vertex value |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user vertex value |
| */ |
| public static <V extends Writable> V createVertexValue(Configuration conf) { |
| Class<V> vertexValueClass = getVertexValueClass(conf); |
| try { |
| return vertexValueClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createVertexValue: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createVertexValue: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Get the user's subclassed edge value class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex edge value class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <E extends Writable> Class<E> getEdgeValueClass(Configuration conf) { |
| return (Class<E>) conf.getClass(PregelixJob.EDGE_VALUE_CLASS, Writable.class); |
| } |
| |
| /** |
| * Create a user edge value |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user edge value |
| */ |
| public static <E extends Writable> E createEdgeValue(Configuration conf) { |
| Class<E> edgeValueClass = getEdgeValueClass(conf); |
| try { |
| return edgeValueClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createEdgeValue: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createEdgeValue: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Get the user's subclassed vertex message value class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's vertex message value class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <M extends WritableSizable> Class<M> getMessageValueClass(Configuration conf) { |
| return (Class<M>) conf.getClass(PregelixJob.MESSAGE_VALUE_CLASS, Writable.class); |
| } |
| |
| /** |
| * Get the user's subclassed global aggregator's partial aggregate value class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's global aggregate value class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <M extends Writable> Class<M> getPartialAggregateValueClass(Configuration conf, String aggClassName) { |
| return (Class<M>) conf.getClass(PregelixJob.PARTIAL_AGGREGATE_VALUE_CLASS + "$" + aggClassName, Writable.class); |
| } |
| |
| /** |
| * Get the user's subclassed combiner's partial combine value class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's global aggregate value class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <M extends Writable> Class<M> getPartialCombineValueClass(Configuration conf) { |
| return (Class<M>) conf.getClass(PregelixJob.PARTIAL_COMBINE_VALUE_CLASS, Writable.class); |
| } |
| |
| /** |
| * Get the user's subclassed normalized key computer class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's normalized key computer class |
| */ |
| @SuppressWarnings("unchecked") |
| public static Class<? extends NormalizedKeyComputer> getNormalizedKeyComputerClass(Configuration conf) { |
| return (Class<? extends NormalizedKeyComputer>) conf.getClass(PregelixJob.NMK_COMPUTER_CLASS, |
| NormalizedKeyComputer.class); |
| } |
| |
| /** |
| * Get the user's subclassed normalized key computer class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return User's global aggregate value class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <M extends Writable> Class<M> getFinalAggregateValueClass(Configuration conf, String aggClassName) { |
| return (Class<M>) conf.getClass(PregelixJob.FINAL_AGGREGATE_VALUE_CLASS + "$" + aggClassName, Writable.class); |
| } |
| |
| /** |
| * Create a user vertex message value |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user vertex message value |
| */ |
| public static <M extends WritableSizable> M createMessageValue(Configuration conf) { |
| Class<M> messageValueClass = getMessageValueClass(conf); |
| try { |
| return messageValueClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createMessageValue: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createMessageValue: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Create a user partial aggregate value |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user aggregate value |
| */ |
| public static <M extends Writable> M createPartialAggregateValue(Configuration conf, String aggClassName) { |
| Class<M> aggregateValueClass = getPartialAggregateValueClass(conf, aggClassName); |
| try { |
| return aggregateValueClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createPartialAggregateValue: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createPartialAggregateValue: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Create the list of user partial aggregate values |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user partial aggregate values |
| */ |
| public static <M extends Writable> List<M> createPartialAggregateValues(Configuration conf) { |
| String[] aggClassNames = BspUtils.getGlobalAggregatorClassNames(conf); |
| List<M> aggValueList = new ArrayList<M>(); |
| for (String aggClassName : aggClassNames) { |
| Class<M> aggregateValueClass = getPartialAggregateValueClass(conf, aggClassName); |
| try { |
| aggValueList.add(aggregateValueClass.newInstance()); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createAggregateValue: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createAggregateValue: Illegally accessed", e); |
| } |
| } |
| return aggValueList; |
| } |
| |
| /** |
| * Create a user partial combine value |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user aggregate value |
| */ |
| @SuppressWarnings("rawtypes") |
| public static <M extends Writable> M createPartialCombineValue(Configuration conf) { |
| Class<M> aggregateValueClass = getPartialCombineValueClass(conf); |
| try { |
| M instance = aggregateValueClass.newInstance(); |
| if (instance instanceof MsgList) { |
| // set conf for msg list, if the value type is msglist |
| ((MsgList) instance).setConf(conf); |
| } |
| return instance; |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createPartialCombineValue: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createPartialCombineValue: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Create a user aggregate value |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user aggregate value |
| */ |
| public static <M extends Writable> M createFinalAggregateValue(Configuration conf, String aggClassName) { |
| Class<M> aggregateValueClass = getFinalAggregateValueClass(conf, aggClassName); |
| try { |
| return aggregateValueClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createAggregateValue: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createAggregateValue: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Create the list of user aggregate values |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user aggregate value |
| */ |
| public static <M extends Writable> List<M> createFinalAggregateValues(Configuration conf) { |
| String[] aggClassNames = BspUtils.getGlobalAggregatorClassNames(conf); |
| List<M> aggValueList = new ArrayList<M>(); |
| for (String aggClassName : aggClassNames) { |
| Class<M> aggregateValueClass = getFinalAggregateValueClass(conf, aggClassName); |
| try { |
| aggValueList.add(aggregateValueClass.newInstance()); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createAggregateValue: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createAggregateValue: Illegally accessed", e); |
| } |
| } |
| return aggValueList; |
| } |
| |
| /** |
| * Create a user aggregate value |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user aggregate value |
| */ |
| @SuppressWarnings("rawtypes") |
| public static VertexPartitioner createVertexPartitioner(Configuration conf) { |
| Class<? extends VertexPartitioner> vertexPartitionerClass = getVertexPartitionerClass(conf); |
| try { |
| return vertexPartitionerClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createVertexPartitioner: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createVertexPartitioner: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Create a checkpoint hook |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user aggregate value |
| */ |
| public static ICheckpointHook createCheckpointHook(Configuration conf) { |
| Class<? extends ICheckpointHook> ckpClass = getCheckpointHookClass(conf); |
| try { |
| return ckpClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createVertexPartitioner: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createVertexPartitioner: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Create a hook that indicates an iteration is complete |
| * |
| * @param conf |
| * Configuration to check |
| * @return Instantiated user aggregate value |
| */ |
| public static IIterationCompleteReporterHook createIterationCompleteHook(Configuration conf) { |
| Class<? extends IIterationCompleteReporterHook> itCompleteClass = getIterationCompleteReporterHookClass(conf); |
| try { |
| return itCompleteClass.newInstance(); |
| } catch (InstantiationException e) { |
| throw new IllegalArgumentException("createVertexPartitioner: Failed to instantiate", e); |
| } catch (IllegalAccessException e) { |
| throw new IllegalArgumentException("createVertexPartitioner: Illegally accessed", e); |
| } |
| } |
| |
| /** |
| * Get the user's subclassed vertex partitioner class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return The user defined vertex partitioner class |
| */ |
| @SuppressWarnings({ "unchecked", "rawtypes" }) |
| public static <V extends VertexPartitioner> Class<V> getVertexPartitionerClass(Configuration conf) { |
| return (Class<V>) conf.getClass(PregelixJob.PARTITIONER_CLASS, null, VertexPartitioner.class); |
| } |
| |
| /** |
| * Get the user's subclassed checkpoint hook class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return The user defined vertex checkpoint hook class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <V extends ICheckpointHook> Class<V> getCheckpointHookClass(Configuration conf) { |
| return (Class<V>) conf.getClass(PregelixJob.CKP_CLASS, DefaultCheckpointHook.class, ICheckpointHook.class); |
| } |
| |
| /** |
| * Get the user's subclassed iteration complete reporter hook class. |
| * |
| * @param conf |
| * Configuration to check |
| * @return The user defined vertex iteration complete reporter class |
| */ |
| @SuppressWarnings("unchecked") |
| public static <V extends IIterationCompleteReporterHook> Class<V> getIterationCompleteReporterHookClass( |
| Configuration conf) { |
| return (Class<V>) conf.getClass(PregelixJob.ITERATION_COMPLETE_CLASS, |
| DefaultIterationCompleteReporterHook.class, IIterationCompleteReporterHook.class); |
| } |
| |
| /** |
| * Get the job configuration parameter whether the vertex states will increase dynamically |
| * |
| * @param conf |
| * the job configuration |
| * @return the boolean setting of the parameter, by default it is false |
| */ |
| public static boolean getDynamicVertexValueSize(Configuration conf) { |
| return conf.getBoolean(PregelixJob.INCREASE_STATE_LENGTH, true); |
| } |
| |
| /** |
| * Get the specified frame size |
| * |
| * @param conf |
| * the job configuration |
| * @return the specified frame size; -1 if it is not set by users |
| */ |
| public static int getFrameSize(Configuration conf) { |
| return conf.getInt(PregelixJob.FRAME_SIZE, -1); |
| } |
| |
| /** |
| * Should the job use LSM or B-tree to store vertices |
| * |
| * @param conf |
| * @return |
| */ |
| public static boolean useLSM(Configuration conf) { |
| return conf.getBoolean(PregelixJob.UPDATE_INTENSIVE, false); |
| } |
| |
| /*** |
| * Get the spilling dir name for global aggregates |
| * |
| * @param conf |
| * @param superStep |
| * @return the spilling dir name |
| */ |
| public static String getGlobalAggregateSpillingDirName(Configuration conf, long superStep) { |
| return TMP_DIR + conf.get(PregelixJob.JOB_ID) + "/agg/" + superStep; |
| } |
| |
| /** |
| * Get the path for vertex checkpointing |
| * |
| * @param conf |
| * @param lastSuperStep |
| * @return the path for vertex checkpointing |
| */ |
| public static String getVertexCheckpointPath(Configuration conf, long lastSuperStep) { |
| return TMP_DIR + BspUtils.getJobId(conf) + "/ckpoint/vertex/" + lastSuperStep; |
| } |
| |
| /** |
| * Get the path for message checkpointing |
| * |
| * @param conf |
| * @param lastSuperStep |
| * @return the path for message checkpointing |
| */ |
| public static String getMessageCheckpointPath(Configuration conf, long lastSuperStep) { |
| String path = TMP_DIR + BspUtils.getJobId(conf) + "/ckpoint/message/" + lastSuperStep; |
| return path; |
| } |
| |
| /** |
| * Get the path for message checkpointing |
| * |
| * @param conf |
| * @param lastSuperStep |
| * @return the path for message checkpointing |
| */ |
| public static String getSecondaryIndexCheckpointPath(Configuration conf, long lastSuperStep) { |
| return TMP_DIR + BspUtils.getJobId(conf) + "/ckpoint/secondaryindex/" + lastSuperStep; |
| } |
| |
| /*** |
| * Get the recovery count |
| * |
| * @return recovery count |
| */ |
| public static int getRecoveryCount(Configuration conf) { |
| return conf.getInt(PregelixJob.RECOVERY_COUNT, 0); |
| } |
| |
| /*** |
| * Get enable dynamic optimization |
| * |
| * @param conf |
| * Configuration |
| * @return true if enabled; otherwise false |
| */ |
| public static boolean getEnableDynamicOptimization(Configuration conf) { |
| return conf.getBoolean(PregelixJob.DYNAMIC_OPTIMIZATION, false); |
| } |
| |
| /*** |
| * Get the user-set checkpoint interval |
| * |
| * @param conf |
| * @return the checkpoint interval |
| */ |
| public static int getCheckpointingInterval(Configuration conf) { |
| return conf.getInt(PregelixJob.CKP_INTERVAL, -1); |
| } |
| |
| /** |
| * Get the grouping algorithm |
| * |
| * @param conf |
| * @return true-sort; false-hash |
| */ |
| public static boolean getGroupingAlgorithm(Configuration conf) { |
| return conf.getBoolean(PregelixJob.GROUPING_ALGORITHM, true); |
| } |
| |
| /** |
| * Get the memory limit for the grouping algorithm (hash only) |
| * |
| * @param conf |
| * @return the memory limit for hash-based grouping |
| */ |
| public static int getGroupingMemoryLimit(Configuration conf) { |
| return conf.getInt(PregelixJob.GROUPING_MEM, 1000); |
| } |
| |
| /** |
| * Get the memory limit for the sort algorithm |
| * |
| * @param conf |
| * @return the memory limit for sorting |
| */ |
| public static int getSortMemoryLimit(Configuration conf) { |
| return conf.getInt(PregelixJob.GROUPING_MEM, 1000); |
| } |
| |
| /** |
| * Get the desired number of workers |
| * |
| * @param conf |
| * @return the number of workers |
| */ |
| public static int getNumberWorkers(Configuration conf) { |
| return conf.getInt(PregelixJob.NUM_WORKERS, -1); |
| } |
| |
| /** |
| * Get whether the combiner key can be skipped when calling a user-defined combine function |
| * |
| * @param conf |
| * @return true to skip; false otherwise |
| */ |
| public static boolean getSkipCombinerKey(Configuration conf) { |
| return conf.getBoolean(PregelixJob.SKIP_COMBINER_KEY, false); |
| } |
| |
| /** |
| * Get whether a merge connector is used |
| * |
| * @param conf |
| * @return true -merge; false-no merge |
| */ |
| public static boolean getMergingConnector(Configuration conf) { |
| return conf.getBoolean(PregelixJob.MERGE_CONNECTOR, true); |
| } |
| |
| /** |
| * return the maximum iteration number |
| * |
| * @param conf |
| * @return the maximum iteration number |
| */ |
| public static int getMaxIteration(Configuration conf) { |
| return conf.getInt(PregelixJob.MAX_ITERATION, Integer.MAX_VALUE); |
| } |
| |
| public static Writable readGlobalAggregateValue(Configuration conf, String jobId, String aggClassName) |
| throws HyracksDataException { |
| try { |
| FileSystem dfs = FileSystem.get(conf); |
| String pathStr = TMP_DIR + jobId + File.separator + "global-agg"; |
| Path path = new Path(pathStr); |
| FSDataInputStream input = dfs.open(path); |
| int numOfAggs = createFinalAggregateValues(conf).size(); |
| for (int i = 0; i < numOfAggs; i++) { |
| String aggName = input.readUTF(); |
| Writable agg = createFinalAggregateValue(conf, aggName); |
| if (aggName.equals(aggClassName)) { |
| agg.readFields(input); |
| input.close(); |
| return agg; |
| } else { |
| agg.readFields(input); |
| } |
| } |
| throw new IllegalStateException("Cannot find the aggregate value for " + aggClassName); |
| } catch (IOException e) { |
| throw new HyracksDataException(e); |
| } |
| } |
| |
| public static HashMap<String, Writable> readAllGlobalAggregateValues(Configuration conf, String jobId) |
| throws HyracksDataException { |
| String pathStr = TMP_DIR + jobId + File.separator + "global-agg"; |
| Path path = new Path(pathStr); |
| List<Writable> aggValues = createFinalAggregateValues(conf); |
| HashMap<String, Writable> finalAggs = new HashMap<>(); |
| try { |
| FileSystem dfs = FileSystem.get(conf); |
| FSDataInputStream input = dfs.open(path); |
| for (int i = 0; i < aggValues.size(); i++) { |
| String aggName = input.readUTF(); |
| aggValues.get(i).readFields(input); |
| finalAggs.put(aggName, aggValues.get(i)); |
| } |
| input.close(); |
| } catch (IOException e) { |
| throw new HyracksDataException(e); |
| } |
| return finalAggs; |
| } |
| |
| public static Counters getCounters(PregelixJob job) throws HyracksDataException { |
| Configuration conf = job.getConfiguration(); |
| String jobId = getJobId(conf); |
| int lastIter = BspUtils.readCountersLastIteration(conf, jobId); |
| return BspUtils.readCounters(lastIter, conf, jobId); |
| } |
| |
| static Counters readCounters(int superstep, Configuration conf, String jobId) throws HyracksDataException { |
| String pathStr = TMP_DIR + jobId + File.separator + BspUtils.COUNTERS_VALUE_ON_ITERATION + superstep; |
| Path path = new Path(pathStr); |
| Counters savedCounters = new Counters(); |
| try { |
| FileSystem dfs = FileSystem.get(conf); |
| FSDataInputStream input = dfs.open(path); |
| savedCounters.readFields(input); |
| input.close(); |
| } catch (IOException e) { |
| throw new HyracksDataException(e); |
| } |
| return savedCounters; |
| } |
| |
| static void writeCounters(Counters toWrite, int superstep, Configuration conf, String jobId) |
| throws HyracksDataException { |
| String pathStr = TMP_DIR + jobId + File.separator + BspUtils.COUNTERS_VALUE_ON_ITERATION + superstep; |
| Path path = new Path(pathStr); |
| try { |
| FileSystem dfs = FileSystem.get(conf); |
| FSDataOutputStream output = dfs.create(path, true); |
| toWrite.write(output); |
| output.close(); |
| } catch (IOException e) { |
| throw new HyracksDataException(e); |
| } |
| } |
| |
| static int readCountersLastIteration(Configuration conf, String jobId) throws HyracksDataException { |
| String pathStr = TMP_DIR + jobId + File.separator + BspUtils.COUNTERS_LAST_ITERATION_COMPLETED; |
| Path path = new Path(pathStr); |
| IntWritable lastIter = new IntWritable(); |
| try { |
| FileSystem dfs = FileSystem.get(conf); |
| FSDataInputStream input = dfs.open(path); |
| lastIter.readFields(input); |
| input.close(); |
| } catch (IOException e) { |
| throw new HyracksDataException(e); |
| } |
| return lastIter.get(); |
| } |
| |
| static void writeCountersLastIteration(int superstep, Configuration conf, String jobId) throws HyracksDataException { |
| String pathStr = TMP_DIR + jobId + File.separator + BspUtils.COUNTERS_LAST_ITERATION_COMPLETED; |
| Path path = new Path(pathStr); |
| try { |
| FileSystem dfs = FileSystem.get(conf); |
| FSDataOutputStream output = dfs.create(path, true); |
| new IntWritable(superstep).write(output); |
| output.close(); |
| } catch (IOException e) { |
| throw new HyracksDataException(e); |
| } |
| } |
| |
| } |
