| /* |
| * 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.conf; |
| |
| import org.apache.giraph.aggregators.AggregatorWriter; |
| import org.apache.giraph.aggregators.TextAggregatorWriter; |
| import org.apache.giraph.bsp.BspOutputFormat; |
| import org.apache.giraph.bsp.checkpoints.CheckpointSupportedChecker; |
| import org.apache.giraph.bsp.checkpoints.DefaultCheckpointSupportedChecker; |
| import org.apache.giraph.combiner.MessageCombiner; |
| import org.apache.giraph.comm.messages.InMemoryMessageStoreFactory; |
| import org.apache.giraph.comm.messages.MessageEncodeAndStoreType; |
| import org.apache.giraph.comm.messages.MessageStoreFactory; |
| import org.apache.giraph.comm.netty.SSLEventHandler; |
| import org.apache.giraph.edge.ByteArrayEdges; |
| import org.apache.giraph.edge.DefaultCreateSourceVertexCallback; |
| import org.apache.giraph.edge.CreateSourceVertexCallback; |
| import org.apache.giraph.edge.EdgeStoreFactory; |
| import org.apache.giraph.edge.InMemoryEdgeStoreFactory; |
| import org.apache.giraph.edge.OutEdges; |
| import org.apache.giraph.factories.ComputationFactory; |
| import org.apache.giraph.factories.DefaultComputationFactory; |
| import org.apache.giraph.factories.DefaultEdgeValueFactory; |
| import org.apache.giraph.factories.DefaultInputOutEdgesFactory; |
| import org.apache.giraph.factories.DefaultMessageValueFactory; |
| import org.apache.giraph.factories.DefaultOutEdgesFactory; |
| import org.apache.giraph.factories.DefaultVertexIdFactory; |
| import org.apache.giraph.factories.DefaultVertexValueFactory; |
| import org.apache.giraph.factories.EdgeValueFactory; |
| import org.apache.giraph.factories.MessageValueFactory; |
| import org.apache.giraph.factories.OutEdgesFactory; |
| import org.apache.giraph.factories.VertexIdFactory; |
| import org.apache.giraph.factories.VertexValueFactory; |
| import org.apache.giraph.graph.Computation; |
| import org.apache.giraph.graph.DefaultVertex; |
| import org.apache.giraph.graph.DefaultVertexResolver; |
| import org.apache.giraph.graph.DefaultVertexValueCombiner; |
| import org.apache.giraph.graph.JobProgressTrackerClient; |
| import org.apache.giraph.graph.Language; |
| import org.apache.giraph.graph.MapperObserver; |
| import org.apache.giraph.graph.RetryableJobProgressTrackerClient; |
| import org.apache.giraph.graph.Vertex; |
| import org.apache.giraph.graph.VertexResolver; |
| import org.apache.giraph.graph.VertexValueCombiner; |
| import org.apache.giraph.io.EdgeInputFormat; |
| import org.apache.giraph.io.EdgeOutputFormat; |
| import org.apache.giraph.io.MappingInputFormat; |
| import org.apache.giraph.io.VertexInputFormat; |
| import org.apache.giraph.io.VertexOutputFormat; |
| import org.apache.giraph.io.filters.DefaultEdgeInputFilter; |
| import org.apache.giraph.io.filters.DefaultVertexInputFilter; |
| import org.apache.giraph.io.filters.EdgeInputFilter; |
| import org.apache.giraph.io.filters.VertexInputFilter; |
| import org.apache.giraph.job.DefaultGiraphJobRetryChecker; |
| import org.apache.giraph.job.DefaultJobObserver; |
| import org.apache.giraph.job.DefaultJobProgressTrackerService; |
| import org.apache.giraph.job.GiraphJobObserver; |
| import org.apache.giraph.job.GiraphJobRetryChecker; |
| import org.apache.giraph.job.HaltApplicationUtils; |
| import org.apache.giraph.job.JobProgressTrackerService; |
| import org.apache.giraph.mapping.MappingStore; |
| import org.apache.giraph.mapping.MappingStoreOps; |
| import org.apache.giraph.mapping.translate.TranslateEdge; |
| import org.apache.giraph.master.DefaultMasterCompute; |
| import org.apache.giraph.master.MasterCompute; |
| import org.apache.giraph.master.MasterObserver; |
| import org.apache.giraph.ooc.persistence.OutOfCoreDataAccessor; |
| import org.apache.giraph.ooc.persistence.LocalDiskDataAccessor; |
| import org.apache.giraph.ooc.policy.MemoryEstimatorOracle; |
| import org.apache.giraph.ooc.policy.OutOfCoreOracle; |
| import org.apache.giraph.partition.GraphPartitionerFactory; |
| import org.apache.giraph.partition.HashPartitionerFactory; |
| import org.apache.giraph.partition.Partition; |
| import org.apache.giraph.partition.SimplePartition; |
| import org.apache.giraph.utils.GcObserver; |
| import org.apache.giraph.worker.DefaultWorkerContext; |
| import org.apache.giraph.worker.WorkerContext; |
| import org.apache.giraph.worker.WorkerObserver; |
| import org.apache.hadoop.io.Writable; |
| import org.apache.hadoop.io.WritableComparable; |
| import org.apache.hadoop.mapreduce.OutputFormat; |
| |
| import static java.util.concurrent.TimeUnit.HOURS; |
| import static java.util.concurrent.TimeUnit.MINUTES; |
| import static java.util.concurrent.TimeUnit.SECONDS; |
| |
| /** |
| * Constants used all over Giraph for configuration. |
| */ |
| // CHECKSTYLE: stop InterfaceIsTypeCheck |
| public interface GiraphConstants { |
| /** 1KB in bytes */ |
| int ONE_KB = 1024; |
| /** 1MB in bytes */ |
| int ONE_MB = 1024 * 1024; |
| |
| /** Mapping related information */ |
| ClassConfOption<MappingStore> MAPPING_STORE_CLASS = |
| ClassConfOption.create("giraph.mappingStoreClass", null, |
| MappingStore.class, "MappingStore Class"); |
| |
| /** Class to use for performing read operations on mapping store */ |
| ClassConfOption<MappingStoreOps> MAPPING_STORE_OPS_CLASS = |
| ClassConfOption.create("giraph.mappingStoreOpsClass", null, |
| MappingStoreOps.class, "MappingStoreOps class"); |
| |
| /** Upper value of LongByteMappingStore */ |
| IntConfOption LB_MAPPINGSTORE_UPPER = |
| new IntConfOption("giraph.lbMappingStoreUpper", -1, |
| "'upper' value used by lbmappingstore"); |
| /** Lower value of LongByteMappingStore */ |
| IntConfOption LB_MAPPINGSTORE_LOWER = |
| new IntConfOption("giraph.lbMappingStoreLower", -1, |
| "'lower' value used by lbMappingstore"); |
| /** Class used to conduct expensive edge translation during vertex input */ |
| ClassConfOption EDGE_TRANSLATION_CLASS = |
| ClassConfOption.create("giraph.edgeTranslationClass", null, |
| TranslateEdge.class, "Class used to conduct expensive edge " + |
| "translation during vertex input phase"); |
| |
| /** Computation class - required */ |
| ClassConfOption<Computation> COMPUTATION_CLASS = |
| ClassConfOption.create("giraph.computationClass", null, |
| Computation.class, "Computation class - required"); |
| /** Computation factory class - optional */ |
| ClassConfOption<ComputationFactory> COMPUTATION_FACTORY_CLASS = |
| ClassConfOption.create("giraph.computation.factory.class", |
| DefaultComputationFactory.class, ComputationFactory.class, |
| "Computation factory class - optional"); |
| |
| /** TypesHolder, used if Computation not set - optional */ |
| ClassConfOption<TypesHolder> TYPES_HOLDER_CLASS = |
| ClassConfOption.create("giraph.typesHolder", null, |
| TypesHolder.class, |
| "TypesHolder, used if Computation not set - optional"); |
| |
| /** Edge Store Factory */ |
| ClassConfOption<EdgeStoreFactory> EDGE_STORE_FACTORY_CLASS = |
| ClassConfOption.create("giraph.edgeStoreFactoryClass", |
| InMemoryEdgeStoreFactory.class, |
| EdgeStoreFactory.class, |
| "Edge Store Factory class to use for creating edgeStore"); |
| |
| /** Message Store Factory */ |
| ClassConfOption<MessageStoreFactory> MESSAGE_STORE_FACTORY_CLASS = |
| ClassConfOption.create("giraph.messageStoreFactoryClass", |
| InMemoryMessageStoreFactory.class, |
| MessageStoreFactory.class, |
| "Message Store Factory Class that is to be used"); |
| |
| /** Language user's graph types are implemented in */ |
| PerGraphTypeEnumConfOption<Language> GRAPH_TYPE_LANGUAGES = |
| PerGraphTypeEnumConfOption.create("giraph.types.language", |
| Language.class, Language.JAVA, |
| "Language user graph types (IVEMM) are implemented in"); |
| |
| /** Whether user graph types need wrappers */ |
| PerGraphTypeBooleanConfOption GRAPH_TYPES_NEEDS_WRAPPERS = |
| new PerGraphTypeBooleanConfOption("giraph.jython.type.wrappers", |
| false, "Whether user graph types (IVEMM) need Jython wrappers"); |
| |
| /** Vertex id factory class - optional */ |
| ClassConfOption<VertexIdFactory> VERTEX_ID_FACTORY_CLASS = |
| ClassConfOption.create("giraph.vertexIdFactoryClass", |
| DefaultVertexIdFactory.class, VertexIdFactory.class, |
| "Vertex ID factory class - optional"); |
| /** Vertex value factory class - optional */ |
| ClassConfOption<VertexValueFactory> VERTEX_VALUE_FACTORY_CLASS = |
| ClassConfOption.create("giraph.vertexValueFactoryClass", |
| DefaultVertexValueFactory.class, VertexValueFactory.class, |
| "Vertex value factory class - optional"); |
| /** Edge value factory class - optional */ |
| ClassConfOption<EdgeValueFactory> EDGE_VALUE_FACTORY_CLASS = |
| ClassConfOption.create("giraph.edgeValueFactoryClass", |
| DefaultEdgeValueFactory.class, EdgeValueFactory.class, |
| "Edge value factory class - optional"); |
| /** Outgoing message value factory class - optional */ |
| ClassConfOption<MessageValueFactory> |
| OUTGOING_MESSAGE_VALUE_FACTORY_CLASS = |
| ClassConfOption.create("giraph.outgoingMessageValueFactoryClass", |
| DefaultMessageValueFactory.class, MessageValueFactory.class, |
| "Outgoing message value factory class - optional"); |
| |
| /** Vertex edges class - optional */ |
| ClassConfOption<OutEdges> VERTEX_EDGES_CLASS = |
| ClassConfOption.create("giraph.outEdgesClass", ByteArrayEdges.class, |
| OutEdges.class, "Vertex edges class - optional"); |
| /** Vertex edges class to be used during edge input only - optional */ |
| ClassConfOption<OutEdges> INPUT_VERTEX_EDGES_CLASS = |
| ClassConfOption.create("giraph.inputOutEdgesClass", |
| ByteArrayEdges.class, OutEdges.class, |
| "Vertex edges class to be used during edge input only - optional"); |
| /** OutEdges factory class - optional */ |
| ClassConfOption<OutEdgesFactory> VERTEX_EDGES_FACTORY_CLASS = |
| ClassConfOption.create("giraph.outEdgesFactoryClass", |
| DefaultOutEdgesFactory.class, OutEdgesFactory.class, |
| "OutEdges factory class - optional"); |
| /** OutEdges for input factory class - optional */ |
| ClassConfOption<OutEdgesFactory> INPUT_VERTEX_EDGES_FACTORY_CLASS = |
| ClassConfOption.create("giraph.inputOutEdgesFactoryClass", |
| DefaultInputOutEdgesFactory.class, OutEdgesFactory.class, |
| "OutEdges for input factory class - optional"); |
| |
| /** Class for Master - optional */ |
| ClassConfOption<MasterCompute> MASTER_COMPUTE_CLASS = |
| ClassConfOption.create("giraph.masterComputeClass", |
| DefaultMasterCompute.class, MasterCompute.class, |
| "Class for Master - optional"); |
| /** Classes for Master Observer - optional */ |
| ClassConfOption<MasterObserver> MASTER_OBSERVER_CLASSES = |
| ClassConfOption.create("giraph.master.observers", |
| null, MasterObserver.class, "Classes for Master Observer - optional"); |
| /** Classes for Worker Observer - optional */ |
| ClassConfOption<WorkerObserver> WORKER_OBSERVER_CLASSES = |
| ClassConfOption.create("giraph.worker.observers", null, |
| WorkerObserver.class, "Classes for Worker Observer - optional"); |
| /** Classes for Mapper Observer - optional */ |
| ClassConfOption<MapperObserver> MAPPER_OBSERVER_CLASSES = |
| ClassConfOption.create("giraph.mapper.observers", null, |
| MapperObserver.class, "Classes for Mapper Observer - optional"); |
| /** Classes for GC Observer - optional */ |
| ClassConfOption<GcObserver> GC_OBSERVER_CLASSES = |
| ClassConfOption.create("giraph.gc.observers", null, |
| GcObserver.class, "Classes for GC oObserver - optional"); |
| /** Message combiner class - optional */ |
| ClassConfOption<MessageCombiner> MESSAGE_COMBINER_CLASS = |
| ClassConfOption.create("giraph.messageCombinerClass", null, |
| MessageCombiner.class, "Message combiner class - optional"); |
| /** Vertex resolver class - optional */ |
| ClassConfOption<VertexResolver> VERTEX_RESOLVER_CLASS = |
| ClassConfOption.create("giraph.vertexResolverClass", |
| DefaultVertexResolver.class, VertexResolver.class, |
| "Vertex resolver class - optional"); |
| /** Vertex value combiner class - optional */ |
| ClassConfOption<VertexValueCombiner> VERTEX_VALUE_COMBINER_CLASS = |
| ClassConfOption.create("giraph.vertexValueCombinerClass", |
| DefaultVertexValueCombiner.class, VertexValueCombiner.class, |
| "Vertex value combiner class - optional"); |
| |
| /** Which language computation is implemented in */ |
| EnumConfOption<Language> COMPUTATION_LANGUAGE = |
| EnumConfOption.create("giraph.computation.language", |
| Language.class, Language.JAVA, |
| "Which language computation is implemented in"); |
| |
| /** |
| * Option of whether to create vertexes that were not existent before but |
| * received messages |
| */ |
| BooleanConfOption RESOLVER_CREATE_VERTEX_ON_MSGS = |
| new BooleanConfOption("giraph.vertex.resolver.create.on.msgs", true, |
| "Option of whether to create vertexes that were not existent " + |
| "before but received messages"); |
| /** Graph partitioner factory class - optional */ |
| ClassConfOption<GraphPartitionerFactory> GRAPH_PARTITIONER_FACTORY_CLASS = |
| ClassConfOption.create("giraph.graphPartitionerFactoryClass", |
| HashPartitionerFactory.class, GraphPartitionerFactory.class, |
| "Graph partitioner factory class - optional"); |
| |
| /** Observer class to watch over job status - optional */ |
| ClassConfOption<GiraphJobObserver> JOB_OBSERVER_CLASS = |
| ClassConfOption.create("giraph.jobObserverClass", |
| DefaultJobObserver.class, GiraphJobObserver.class, |
| "Observer class to watch over job status - optional"); |
| |
| /** Observer class to watch over job status - optional */ |
| ClassConfOption<GiraphJobRetryChecker> JOB_RETRY_CHECKER_CLASS = |
| ClassConfOption.create("giraph.jobRetryCheckerClass", |
| DefaultGiraphJobRetryChecker.class, GiraphJobRetryChecker.class, |
| "Class which decides whether a failed job should be retried - " + |
| "optional"); |
| |
| /** |
| * Maximum allowed time for job to run after getting all resources before it |
| * will be killed, in milliseconds (-1 if it has no limit) |
| */ |
| LongConfOption MAX_ALLOWED_JOB_TIME_MS = |
| new LongConfOption("giraph.maxAllowedJobTimeMilliseconds", -1, |
| "Maximum allowed time for job to run after getting all resources " + |
| "before it will be killed, in milliseconds " + |
| "(-1 if it has no limit)"); |
| |
| // At least one of the input format classes is required. |
| /** VertexInputFormat class */ |
| ClassConfOption<VertexInputFormat> VERTEX_INPUT_FORMAT_CLASS = |
| ClassConfOption.create("giraph.vertexInputFormatClass", null, |
| VertexInputFormat.class, "VertexInputFormat class (at least " + |
| "one of the input format classes is required)"); |
| /** EdgeInputFormat class */ |
| ClassConfOption<EdgeInputFormat> EDGE_INPUT_FORMAT_CLASS = |
| ClassConfOption.create("giraph.edgeInputFormatClass", null, |
| EdgeInputFormat.class, "EdgeInputFormat class"); |
| /** MappingInputFormat class */ |
| ClassConfOption<MappingInputFormat> MAPPING_INPUT_FORMAT_CLASS = |
| ClassConfOption.create("giraph.mappingInputFormatClass", null, |
| MappingInputFormat.class, "MappingInputFormat class"); |
| |
| /** EdgeInputFilter class */ |
| ClassConfOption<EdgeInputFilter> EDGE_INPUT_FILTER_CLASS = |
| ClassConfOption.create("giraph.edgeInputFilterClass", |
| DefaultEdgeInputFilter.class, EdgeInputFilter.class, |
| "EdgeInputFilter class"); |
| /** VertexInputFilter class */ |
| ClassConfOption<VertexInputFilter> VERTEX_INPUT_FILTER_CLASS = |
| ClassConfOption.create("giraph.vertexInputFilterClass", |
| DefaultVertexInputFilter.class, VertexInputFilter.class, |
| "VertexInputFilter class"); |
| /** Vertex class */ |
| ClassConfOption<Vertex> VERTEX_CLASS = |
| ClassConfOption.create("giraph.vertexClass", |
| DefaultVertex.class, Vertex.class, |
| "Vertex class"); |
| /** VertexOutputFormat class */ |
| ClassConfOption<VertexOutputFormat> VERTEX_OUTPUT_FORMAT_CLASS = |
| ClassConfOption.create("giraph.vertexOutputFormatClass", null, |
| VertexOutputFormat.class, "VertexOutputFormat class"); |
| /** EdgeOutputFormat sub-directory */ |
| StrConfOption VERTEX_OUTPUT_FORMAT_SUBDIR = |
| new StrConfOption("giraph.vertex.output.subdir", "", |
| "VertexOutputFormat sub-directory"); |
| /** EdgeOutputFormat class */ |
| ClassConfOption<EdgeOutputFormat> EDGE_OUTPUT_FORMAT_CLASS = |
| ClassConfOption.create("giraph.edgeOutputFormatClass", null, |
| EdgeOutputFormat.class, "EdgeOutputFormat class"); |
| /** EdgeOutputFormat sub-directory */ |
| StrConfOption EDGE_OUTPUT_FORMAT_SUBDIR = |
| new StrConfOption("giraph.edge.output.subdir", "", |
| "EdgeOutputFormat sub-directory"); |
| |
| /** GiraphTextOuputFormat Separator */ |
| StrConfOption GIRAPH_TEXT_OUTPUT_FORMAT_SEPARATOR = |
| new StrConfOption("giraph.textoutputformat.separator", "\t", |
| "GiraphTextOuputFormat Separator"); |
| /** Reverse values in the output */ |
| BooleanConfOption GIRAPH_TEXT_OUTPUT_FORMAT_REVERSE = |
| new BooleanConfOption("giraph.textoutputformat.reverse", false, |
| "Reverse values in the output"); |
| |
| /** |
| * If you use this option, instead of having saving vertices in the end of |
| * application, saveVertex will be called right after each vertex.compute() |
| * is called. |
| * NOTE: This feature doesn't work well with checkpointing - if you restart |
| * from a checkpoint you won't have any output from previous supersteps. |
| */ |
| BooleanConfOption DO_OUTPUT_DURING_COMPUTATION = |
| new BooleanConfOption("giraph.doOutputDuringComputation", false, |
| "If you use this option, instead of having saving vertices in the " + |
| "end of application, saveVertex will be called right after each " + |
| "vertex.compute() is called." + |
| "NOTE: This feature doesn't work well with checkpointing - if you " + |
| "restart from a checkpoint you won't have any ouptut from previous " + |
| "supresteps."); |
| /** |
| * Vertex output format thread-safe - if your VertexOutputFormat allows |
| * several vertexWriters to be created and written to in parallel, |
| * you should set this to true. |
| */ |
| BooleanConfOption VERTEX_OUTPUT_FORMAT_THREAD_SAFE = |
| new BooleanConfOption("giraph.vertexOutputFormatThreadSafe", false, |
| "Vertex output format thread-safe - if your VertexOutputFormat " + |
| "allows several vertexWriters to be created and written to in " + |
| "parallel, you should set this to true."); |
| /** Number of threads for writing output in the end of the application */ |
| IntConfOption NUM_OUTPUT_THREADS = |
| new IntConfOption("giraph.numOutputThreads", 1, |
| "Number of threads for writing output in the end of the application"); |
| |
| /** conf key for comma-separated list of jars to export to YARN workers */ |
| StrConfOption GIRAPH_YARN_LIBJARS = |
| new StrConfOption("giraph.yarn.libjars", "", |
| "conf key for comma-separated list of jars to export to YARN workers"); |
| /** Name of the XML file that will export our Configuration to YARN workers */ |
| String GIRAPH_YARN_CONF_FILE = "giraph-conf.xml"; |
| /** Giraph default heap size for all tasks when running on YARN profile */ |
| int GIRAPH_YARN_TASK_HEAP_MB_DEFAULT = 1024; |
| /** Name of Giraph property for user-configurable heap memory per worker */ |
| IntConfOption GIRAPH_YARN_TASK_HEAP_MB = new IntConfOption( |
| "giraph.yarn.task.heap.mb", GIRAPH_YARN_TASK_HEAP_MB_DEFAULT, |
| "Name of Giraph property for user-configurable heap memory per worker"); |
| /** Default priority level in YARN for our task containers */ |
| int GIRAPH_YARN_PRIORITY = 10; |
| /** Is this a pure YARN job (i.e. no MapReduce layer managing Giraph tasks) */ |
| BooleanConfOption IS_PURE_YARN_JOB = |
| new BooleanConfOption("giraph.pure.yarn.job", false, |
| "Is this a pure YARN job (i.e. no MapReduce layer managing Giraph " + |
| "tasks)"); |
| |
| /** Vertex index class */ |
| ClassConfOption<WritableComparable> VERTEX_ID_CLASS = |
| ClassConfOption.create("giraph.vertexIdClass", null, |
| WritableComparable.class, "Vertex index class"); |
| /** Vertex value class */ |
| ClassConfOption<Writable> VERTEX_VALUE_CLASS = |
| ClassConfOption.create("giraph.vertexValueClass", null, Writable.class, |
| "Vertex value class"); |
| /** Edge value class */ |
| ClassConfOption<Writable> EDGE_VALUE_CLASS = |
| ClassConfOption.create("giraph.edgeValueClass", null, Writable.class, |
| "Edge value class"); |
| /** Outgoing message value class */ |
| ClassConfOption<Writable> OUTGOING_MESSAGE_VALUE_CLASS = |
| ClassConfOption.create("giraph.outgoingMessageValueClass", null, |
| Writable.class, "Outgoing message value class"); |
| /** Worker context class */ |
| ClassConfOption<WorkerContext> WORKER_CONTEXT_CLASS = |
| ClassConfOption.create("giraph.workerContextClass", |
| DefaultWorkerContext.class, WorkerContext.class, |
| "Worker contextclass"); |
| /** AggregatorWriter class - optional */ |
| ClassConfOption<AggregatorWriter> AGGREGATOR_WRITER_CLASS = |
| ClassConfOption.create("giraph.aggregatorWriterClass", |
| TextAggregatorWriter.class, AggregatorWriter.class, |
| "AggregatorWriter class - optional"); |
| |
| /** Partition class - optional */ |
| ClassConfOption<Partition> PARTITION_CLASS = |
| ClassConfOption.create("giraph.partitionClass", SimplePartition.class, |
| Partition.class, "Partition class - optional"); |
| |
| /** |
| * Minimum number of simultaneous workers before this job can run (int) |
| */ |
| String MIN_WORKERS = "giraph.minWorkers"; |
| /** |
| * Maximum number of simultaneous worker tasks started by this job (int). |
| */ |
| String MAX_WORKERS = "giraph.maxWorkers"; |
| |
| /** |
| * Separate the workers and the master tasks. This is required |
| * to support dynamic recovery. (boolean) |
| */ |
| BooleanConfOption SPLIT_MASTER_WORKER = |
| new BooleanConfOption("giraph.SplitMasterWorker", true, |
| "Separate the workers and the master tasks. This is required to " + |
| "support dynamic recovery. (boolean)"); |
| |
| /** Indicates whether this job is run in an internal unit test */ |
| BooleanConfOption LOCAL_TEST_MODE = |
| new BooleanConfOption("giraph.localTestMode", false, |
| "Indicates whether this job is run in an internal unit test"); |
| |
| /** Override the Hadoop log level and set the desired log level. */ |
| StrConfOption LOG_LEVEL = new StrConfOption("giraph.logLevel", "info", |
| "Override the Hadoop log level and set the desired log level."); |
| |
| /** Use thread level debugging? */ |
| BooleanConfOption LOG_THREAD_LAYOUT = |
| new BooleanConfOption("giraph.logThreadLayout", false, |
| "Use thread level debugging?"); |
| |
| /** Configuration key to enable jmap printing */ |
| BooleanConfOption JMAP_ENABLE = |
| new BooleanConfOption("giraph.jmap.histo.enable", false, |
| "Configuration key to enable jmap printing"); |
| |
| /** Configuration key for msec to sleep between calls */ |
| IntConfOption JMAP_SLEEP_MILLIS = |
| new IntConfOption("giraph.jmap.histo.msec", SECONDS.toMillis(30), |
| "Configuration key for msec to sleep between calls"); |
| |
| /** Configuration key for how many lines to print */ |
| IntConfOption JMAP_PRINT_LINES = |
| new IntConfOption("giraph.jmap.histo.print_lines", 30, |
| "Configuration key for how many lines to print"); |
| |
| /** |
| * Configuration key for printing live objects only |
| * This option will trigger Full GC for every jmap dump |
| * and so can significantly hinder performance. |
| */ |
| BooleanConfOption JMAP_LIVE_ONLY = |
| new BooleanConfOption("giraph.jmap.histo.live", false, |
| "Only print live objects in jmap?"); |
| |
| /** |
| * Option used by ReactiveJMapHistoDumper to check for an imminent |
| * OOM in worker or master process |
| */ |
| IntConfOption MIN_FREE_MBS_ON_HEAP = |
| new IntConfOption("giraph.heap.minFreeMb", 128, "Option used by " + |
| "worker and master observers to check for imminent OOM exception"); |
| /** |
| * Option can be used to enable reactively dumping jmap histo when |
| * OOM is imminent |
| */ |
| BooleanConfOption REACTIVE_JMAP_ENABLE = |
| new BooleanConfOption("giraph.heap.enableReactiveJmapDumping", false, |
| "Option to enable dumping jmap histogram reactively based on " + |
| "free memory on heap"); |
| |
| /** |
| * Minimum percent of the maximum number of workers that have responded |
| * in order to continue progressing. (float) |
| */ |
| FloatConfOption MIN_PERCENT_RESPONDED = |
| new FloatConfOption("giraph.minPercentResponded", 100.0f, |
| "Minimum percent of the maximum number of workers that have " + |
| "responded in order to continue progressing. (float)"); |
| |
| /** Enable the Metrics system */ |
| BooleanConfOption METRICS_ENABLE = |
| new BooleanConfOption("giraph.metrics.enable", false, |
| "Enable the Metrics system"); |
| |
| /** Directory in HDFS to write master metrics to, instead of stderr */ |
| StrConfOption METRICS_DIRECTORY = |
| new StrConfOption("giraph.metrics.directory", "", |
| "Directory in HDFS to write master metrics to, instead of stderr"); |
| |
| /** |
| * ZooKeeper comma-separated list (if not set, |
| * will start up ZooKeeper locally). Consider that after locally-starting |
| * zookeeper, this parameter will updated the configuration with the corrent |
| * configuration value. |
| */ |
| StrConfOption ZOOKEEPER_LIST = |
| new StrConfOption("giraph.zkList", "", |
| "ZooKeeper comma-separated list (if not set, will start up " + |
| "ZooKeeper locally). Consider that after locally-starting " + |
| "zookeeper, this parameter will updated the configuration with " + |
| "the corrent configuration value."); |
| |
| /** |
| * Zookeeper List will always hold a value during the computation while |
| * this option provides information regarding whether the zookeeper was |
| * internally started or externally provided. |
| */ |
| BooleanConfOption ZOOKEEPER_IS_EXTERNAL = |
| new BooleanConfOption("giraph.zkIsExternal", true, |
| "Zookeeper List will always hold a value during " + |
| "the computation while this option provides " + |
| "information regarding whether the zookeeper was " + |
| "internally started or externally provided."); |
| |
| /** ZooKeeper session millisecond timeout */ |
| IntConfOption ZOOKEEPER_SESSION_TIMEOUT = |
| new IntConfOption("giraph.zkSessionMsecTimeout", MINUTES.toMillis(1), |
| "ZooKeeper session millisecond timeout"); |
| |
| /** Polling interval to check for the ZooKeeper server data */ |
| IntConfOption ZOOKEEPER_SERVERLIST_POLL_MSECS = |
| new IntConfOption("giraph.zkServerlistPollMsecs", SECONDS.toMillis(3), |
| "Polling interval to check for the ZooKeeper server data"); |
| |
| /** ZooKeeper port to use */ |
| IntConfOption ZOOKEEPER_SERVER_PORT = |
| new IntConfOption("giraph.zkServerPort", 22181, "ZooKeeper port to use"); |
| |
| /** Local ZooKeeper directory to use */ |
| String ZOOKEEPER_DIR = "giraph.zkDir"; |
| |
| /** Max attempts for handling ZooKeeper connection loss */ |
| IntConfOption ZOOKEEPER_OPS_MAX_ATTEMPTS = |
| new IntConfOption("giraph.zkOpsMaxAttempts", 3, |
| "Max attempts for handling ZooKeeper connection loss"); |
| |
| /** |
| * Msecs to wait before retrying a failed ZooKeeper op due to connection loss. |
| */ |
| IntConfOption ZOOKEEPER_OPS_RETRY_WAIT_MSECS = |
| new IntConfOption("giraph.zkOpsRetryWaitMsecs", SECONDS.toMillis(5), |
| "Msecs to wait before retrying a failed ZooKeeper op due to " + |
| "connection loss."); |
| |
| /** TCP backlog (defaults to number of workers) */ |
| IntConfOption TCP_BACKLOG = new IntConfOption("giraph.tcpBacklog", 1, |
| "TCP backlog (defaults to number of workers)"); |
| |
| /** Use netty pooled memory buffer allocator */ |
| BooleanConfOption NETTY_USE_POOLED_ALLOCATOR = new BooleanConfOption( |
| "giraph.useNettyPooledAllocator", false, "Should netty use pooled " + |
| "memory allocator?"); |
| |
| /** Use direct memory buffers in netty */ |
| BooleanConfOption NETTY_USE_DIRECT_MEMORY = new BooleanConfOption( |
| "giraph.useNettyDirectMemory", false, "Should netty use direct " + |
| "memory buffers"); |
| |
| /** How big to make the encoder buffer? */ |
| IntConfOption NETTY_REQUEST_ENCODER_BUFFER_SIZE = |
| new IntConfOption("giraph.nettyRequestEncoderBufferSize", 32 * ONE_KB, |
| "How big to make the encoder buffer?"); |
| |
| /** Netty client threads */ |
| IntConfOption NETTY_CLIENT_THREADS = |
| new IntConfOption("giraph.nettyClientThreads", 4, "Netty client threads"); |
| |
| /** Netty server threads */ |
| IntConfOption NETTY_SERVER_THREADS = |
| new IntConfOption("giraph.nettyServerThreads", 16, |
| "Netty server threads"); |
| |
| /** Use the execution handler in netty on the client? */ |
| BooleanConfOption NETTY_CLIENT_USE_EXECUTION_HANDLER = |
| new BooleanConfOption("giraph.nettyClientUseExecutionHandler", true, |
| "Use the execution handler in netty on the client?"); |
| |
| /** Netty client execution threads (execution handler) */ |
| IntConfOption NETTY_CLIENT_EXECUTION_THREADS = |
| new IntConfOption("giraph.nettyClientExecutionThreads", 8, |
| "Netty client execution threads (execution handler)"); |
| |
| /** Where to place the netty client execution handle? */ |
| StrConfOption NETTY_CLIENT_EXECUTION_AFTER_HANDLER = |
| new StrConfOption("giraph.nettyClientExecutionAfterHandler", |
| "request-encoder", |
| "Where to place the netty client execution handle?"); |
| |
| /** Use the execution handler in netty on the server? */ |
| BooleanConfOption NETTY_SERVER_USE_EXECUTION_HANDLER = |
| new BooleanConfOption("giraph.nettyServerUseExecutionHandler", true, |
| "Use the execution handler in netty on the server?"); |
| |
| /** Netty server execution threads (execution handler) */ |
| IntConfOption NETTY_SERVER_EXECUTION_THREADS = |
| new IntConfOption("giraph.nettyServerExecutionThreads", 8, |
| "Netty server execution threads (execution handler)"); |
| |
| /** Where to place the netty server execution handle? */ |
| StrConfOption NETTY_SERVER_EXECUTION_AFTER_HANDLER = |
| new StrConfOption("giraph.nettyServerExecutionAfterHandler", |
| "requestFrameDecoder", |
| "Where to place the netty server execution handle?"); |
| |
| /** Netty simulate a first request closed */ |
| BooleanConfOption NETTY_SIMULATE_FIRST_REQUEST_CLOSED = |
| new BooleanConfOption("giraph.nettySimulateFirstRequestClosed", false, |
| "Netty simulate a first request closed"); |
| |
| /** Netty simulate a first response failed */ |
| BooleanConfOption NETTY_SIMULATE_FIRST_RESPONSE_FAILED = |
| new BooleanConfOption("giraph.nettySimulateFirstResponseFailed", false, |
| "Netty simulate a first response failed"); |
| |
| /** Netty - set which compression to use */ |
| StrConfOption NETTY_COMPRESSION_ALGORITHM = |
| new StrConfOption("giraph.nettyCompressionAlgorithm", "", |
| "Which compression algorithm to use in netty"); |
| |
| /** Max resolve address attempts */ |
| IntConfOption MAX_RESOLVE_ADDRESS_ATTEMPTS = |
| new IntConfOption("giraph.maxResolveAddressAttempts", 5, |
| "Max resolve address attempts"); |
| |
| /** Msecs to wait between waiting for all requests to finish */ |
| IntConfOption WAITING_REQUEST_MSECS = |
| new IntConfOption("giraph.waitingRequestMsecs", SECONDS.toMillis(15), |
| "Msecs to wait between waiting for all requests to finish"); |
| |
| /** Millseconds to wait for an event before continuing */ |
| IntConfOption EVENT_WAIT_MSECS = |
| new IntConfOption("giraph.eventWaitMsecs", SECONDS.toMillis(30), |
| "Millseconds to wait for an event before continuing"); |
| |
| /** |
| * Maximum milliseconds to wait before giving up trying to get the minimum |
| * number of workers before a superstep (int). |
| */ |
| IntConfOption MAX_MASTER_SUPERSTEP_WAIT_MSECS = |
| new IntConfOption("giraph.maxMasterSuperstepWaitMsecs", |
| MINUTES.toMillis(10), |
| "Maximum milliseconds to wait before giving up trying to get the " + |
| "minimum number of workers before a superstep (int)."); |
| |
| /** |
| * Maximum milliseconds to wait before giving up waiting for the workers to |
| * write the counters to the Zookeeper after a superstep |
| */ |
| IntConfOption MAX_COUNTER_WAIT_MSECS = new IntConfOption( |
| "giraph.maxCounterWaitMsecs", MINUTES.toMillis(2), |
| "Maximum milliseconds to wait before giving up waiting for" + |
| "the workers to write their counters to the " + |
| "zookeeper after a superstep"); |
| |
| /** Milliseconds for a request to complete (or else resend) */ |
| IntConfOption MAX_REQUEST_MILLISECONDS = |
| new IntConfOption("giraph.maxRequestMilliseconds", MINUTES.toMillis(10), |
| "Milliseconds for a request to complete (or else resend)"); |
| |
| /** |
| * Whether to resend request which timed out or fail the job if timeout |
| * happens |
| */ |
| BooleanConfOption RESEND_TIMED_OUT_REQUESTS = |
| new BooleanConfOption("giraph.resendTimedOutRequests", true, |
| "Whether to resend request which timed out or fail the job if " + |
| "timeout happens"); |
| |
| /** Netty max connection failures */ |
| IntConfOption NETTY_MAX_CONNECTION_FAILURES = |
| new IntConfOption("giraph.nettyMaxConnectionFailures", 1000, |
| "Netty max connection failures"); |
| |
| /** How long to wait before trying to reconnect failed connections */ |
| IntConfOption WAIT_TIME_BETWEEN_CONNECTION_RETRIES_MS = |
| new IntConfOption("giraph.waitTimeBetweenConnectionRetriesMs", 500, |
| ""); |
| |
| /** Initial port to start using for the IPC communication */ |
| IntConfOption IPC_INITIAL_PORT = |
| new IntConfOption("giraph.ipcInitialPort", 30000, |
| "Initial port to start using for the IPC communication"); |
| |
| /** Maximum bind attempts for different IPC ports */ |
| IntConfOption MAX_IPC_PORT_BIND_ATTEMPTS = |
| new IntConfOption("giraph.maxIpcPortBindAttempts", 20, |
| "Maximum bind attempts for different IPC ports"); |
| /** |
| * Fail first IPC port binding attempt, simulate binding failure |
| * on real grid testing |
| */ |
| BooleanConfOption FAIL_FIRST_IPC_PORT_BIND_ATTEMPT = |
| new BooleanConfOption("giraph.failFirstIpcPortBindAttempt", false, |
| "Fail first IPC port binding attempt, simulate binding failure " + |
| "on real grid testing"); |
| |
| /** Client send buffer size */ |
| IntConfOption CLIENT_SEND_BUFFER_SIZE = |
| new IntConfOption("giraph.clientSendBufferSize", 512 * ONE_KB, |
| "Client send buffer size"); |
| |
| /** Client receive buffer size */ |
| IntConfOption CLIENT_RECEIVE_BUFFER_SIZE = |
| new IntConfOption("giraph.clientReceiveBufferSize", 32 * ONE_KB, |
| "Client receive buffer size"); |
| |
| /** Server send buffer size */ |
| IntConfOption SERVER_SEND_BUFFER_SIZE = |
| new IntConfOption("giraph.serverSendBufferSize", 32 * ONE_KB, |
| "Server send buffer size"); |
| |
| /** Server receive buffer size */ |
| IntConfOption SERVER_RECEIVE_BUFFER_SIZE = |
| new IntConfOption("giraph.serverReceiveBufferSize", 512 * ONE_KB, |
| "Server receive buffer size"); |
| |
| /** Maximum size of messages (in bytes) per peer before flush */ |
| IntConfOption MAX_MSG_REQUEST_SIZE = |
| new IntConfOption("giraph.msgRequestSize", 512 * ONE_KB, |
| "Maximum size of messages (in bytes) per peer before flush"); |
| |
| /** |
| * How much bigger than the average per partition size to make initial per |
| * partition buffers. |
| * If this value is A, message request size is M, |
| * and a worker has P partitions, than its initial partition buffer size |
| * will be (M / P) * (1 + A). |
| */ |
| FloatConfOption ADDITIONAL_MSG_REQUEST_SIZE = |
| new FloatConfOption("giraph.additionalMsgRequestSize", 0.2f, |
| "How much bigger than the average per partition size to make " + |
| "initial per partition buffers. If this value is A, message " + |
| "request size is M, and a worker has P partitions, than its " + |
| "initial partition buffer size will be (M / P) * (1 + A)."); |
| |
| |
| /** Warn if msg request size exceeds default size by this factor */ |
| FloatConfOption REQUEST_SIZE_WARNING_THRESHOLD = new FloatConfOption( |
| "giraph.msgRequestWarningThreshold", 2.0f, |
| "If request sizes are bigger than the buffer size by this factor " + |
| "warnings are printed to the log and to the command line"); |
| |
| /** Maximum size of vertices (in bytes) per peer before flush */ |
| IntConfOption MAX_VERTEX_REQUEST_SIZE = |
| new IntConfOption("giraph.vertexRequestSize", 512 * ONE_KB, |
| "Maximum size of vertices (in bytes) per peer before flush"); |
| |
| /** |
| * Additional size (expressed as a ratio) of each per-partition buffer on |
| * top of the average size for vertices. |
| */ |
| FloatConfOption ADDITIONAL_VERTEX_REQUEST_SIZE = |
| new FloatConfOption("giraph.additionalVertexRequestSize", 0.2f, |
| "Additional size (expressed as a ratio) of each per-partition " + |
| "buffer on top of the average size."); |
| |
| /** Maximum size of edges (in bytes) per peer before flush */ |
| IntConfOption MAX_EDGE_REQUEST_SIZE = |
| new IntConfOption("giraph.edgeRequestSize", 512 * ONE_KB, |
| "Maximum size of edges (in bytes) per peer before flush"); |
| |
| /** |
| * Additional size (expressed as a ratio) of each per-partition buffer on |
| * top of the average size for edges. |
| */ |
| FloatConfOption ADDITIONAL_EDGE_REQUEST_SIZE = |
| new FloatConfOption("giraph.additionalEdgeRequestSize", 0.2f, |
| "Additional size (expressed as a ratio) of each per-partition " + |
| "buffer on top of the average size."); |
| |
| /** Maximum number of mutations per partition before flush */ |
| IntConfOption MAX_MUTATIONS_PER_REQUEST = |
| new IntConfOption("giraph.maxMutationsPerRequest", 100, |
| "Maximum number of mutations per partition before flush"); |
| |
| /** |
| * Use message size encoding (typically better for complex objects, |
| * not meant for primitive wrapped messages) |
| */ |
| BooleanConfOption USE_MESSAGE_SIZE_ENCODING = |
| new BooleanConfOption("giraph.useMessageSizeEncoding", false, |
| "Use message size encoding (typically better for complex objects, " + |
| "not meant for primitive wrapped messages)"); |
| |
| /** Number of channels used per server */ |
| IntConfOption CHANNELS_PER_SERVER = |
| new IntConfOption("giraph.channelsPerServer", 1, |
| "Number of channels used per server"); |
| |
| /** Number of flush threads per peer */ |
| String MSG_NUM_FLUSH_THREADS = "giraph.msgNumFlushThreads"; |
| |
| /** Number of threads for vertex computation */ |
| IntConfOption NUM_COMPUTE_THREADS = |
| new IntConfOption("giraph.numComputeThreads", 1, |
| "Number of threads for vertex computation"); |
| |
| /** Number of threads for input split loading */ |
| IntConfOption NUM_INPUT_THREADS = |
| new IntConfOption("giraph.numInputThreads", 1, |
| "Number of threads for input split loading"); |
| |
| /** Minimum stragglers of the superstep before printing them out */ |
| IntConfOption PARTITION_LONG_TAIL_MIN_PRINT = |
| new IntConfOption("giraph.partitionLongTailMinPrint", 1, |
| "Minimum stragglers of the superstep before printing them out"); |
| |
| /** Use superstep counters? (boolean) */ |
| BooleanConfOption USE_SUPERSTEP_COUNTERS = |
| new BooleanConfOption("giraph.useSuperstepCounters", true, |
| "Use superstep counters? (boolean)"); |
| |
| /** |
| * Input split sample percent - Used only for sampling and testing, rather |
| * than an actual job. The idea is that to test, you might only want a |
| * fraction of the actual input splits from your VertexInputFormat to |
| * load (values should be [0, 100]). |
| */ |
| FloatConfOption INPUT_SPLIT_SAMPLE_PERCENT = |
| new FloatConfOption("giraph.inputSplitSamplePercent", 100f, |
| "Input split sample percent - Used only for sampling and testing, " + |
| "rather than an actual job. The idea is that to test, you might " + |
| "only want a fraction of the actual input splits from your " + |
| "VertexInputFormat to load (values should be [0, 100])."); |
| |
| /** |
| * To limit outlier vertex input splits from producing too many vertices or |
| * to help with testing, the number of vertices loaded from an input split |
| * can be limited. By default, everything is loaded. |
| */ |
| LongConfOption INPUT_SPLIT_MAX_VERTICES = |
| new LongConfOption("giraph.InputSplitMaxVertices", -1, |
| "To limit outlier vertex input splits from producing too many " + |
| "vertices or to help with testing, the number of vertices " + |
| "loaded from an input split can be limited. By default, " + |
| "everything is loaded."); |
| |
| /** |
| * To limit outlier vertex input splits from producing too many vertices or |
| * to help with testing, the number of edges loaded from an input split |
| * can be limited. By default, everything is loaded. |
| */ |
| LongConfOption INPUT_SPLIT_MAX_EDGES = |
| new LongConfOption("giraph.InputSplitMaxEdges", -1, |
| "To limit outlier vertex input splits from producing too many " + |
| "vertices or to help with testing, the number of edges loaded " + |
| "from an input split can be limited. By default, everything is " + |
| "loaded."); |
| |
| /** |
| * To minimize network usage when reading input splits, |
| * each worker can prioritize splits that reside on its host. |
| * This, however, comes at the cost of increased load on ZooKeeper. |
| * Hence, users with a lot of splits and input threads (or with |
| * configurations that can't exploit locality) may want to disable it. |
| */ |
| BooleanConfOption USE_INPUT_SPLIT_LOCALITY = |
| new BooleanConfOption("giraph.useInputSplitLocality", true, |
| "To minimize network usage when reading input splits, each worker " + |
| "can prioritize splits that reside on its host. " + |
| "This, however, comes at the cost of increased load on ZooKeeper. " + |
| "Hence, users with a lot of splits and input threads (or with " + |
| "configurations that can't exploit locality) may want to disable " + |
| "it."); |
| |
| /** Multiplier for the current workers squared */ |
| FloatConfOption PARTITION_COUNT_MULTIPLIER = |
| new FloatConfOption("giraph.masterPartitionCountMultiplier", 1.0f, |
| "Multiplier for the current workers squared"); |
| |
| /** Minimum number of partitions to have per compute thread */ |
| IntConfOption MIN_PARTITIONS_PER_COMPUTE_THREAD = |
| new IntConfOption("giraph.minPartitionsPerComputeThread", 1, |
| "Minimum number of partitions to have per compute thread"); |
| |
| /** Overrides default partition count calculation if not -1 */ |
| IntConfOption USER_PARTITION_COUNT = |
| new IntConfOption("giraph.userPartitionCount", -1, |
| "Overrides default partition count calculation if not -1"); |
| |
| /** Vertex key space size for |
| * {@link org.apache.giraph.partition.WorkerGraphPartitionerImpl} |
| */ |
| String PARTITION_VERTEX_KEY_SPACE_SIZE = "giraph.vertexKeySpaceSize"; |
| |
| /** |
| * How often to checkpoint (i.e. 0, means no checkpoint, |
| * 1 means every superstep, 2 is every two supersteps, etc.). |
| */ |
| IntConfOption CHECKPOINT_FREQUENCY = |
| new IntConfOption("giraph.checkpointFrequency", 0, |
| "How often to checkpoint (i.e. 0, means no checkpoint, 1 means " + |
| "every superstep, 2 is every two supersteps, etc.)."); |
| |
| /** |
| * Delete checkpoints after a successful job run? |
| */ |
| BooleanConfOption CLEANUP_CHECKPOINTS_AFTER_SUCCESS = |
| new BooleanConfOption("giraph.cleanupCheckpointsAfterSuccess", true, |
| "Delete checkpoints after a successful job run?"); |
| |
| /** |
| * An application can be restarted manually by selecting a superstep. The |
| * corresponding checkpoint must exist for this to work. The user should |
| * set a long value. Default is start from scratch. |
| */ |
| String RESTART_SUPERSTEP = "giraph.restartSuperstep"; |
| |
| /** |
| * If application is restarted manually we need to specify job ID |
| * to restart from. |
| */ |
| StrConfOption RESTART_JOB_ID = new StrConfOption("giraph.restart.jobId", |
| null, "Which job ID should I try to restart?"); |
| |
| /** |
| * Base ZNode for Giraph's state in the ZooKeeper cluster. Must be a root |
| * znode on the cluster beginning with "/" |
| */ |
| String BASE_ZNODE_KEY = "giraph.zkBaseZNode"; |
| |
| /** |
| * If ZOOKEEPER_LIST is not set, then use this directory to manage |
| * ZooKeeper |
| */ |
| StrConfOption ZOOKEEPER_MANAGER_DIRECTORY = |
| new StrConfOption("giraph.zkManagerDirectory", |
| "_bsp/_defaultZkManagerDir", |
| "If ZOOKEEPER_LIST is not set, then use this directory to manage " + |
| "ZooKeeper"); |
| |
| /** Number of ZooKeeper client connection attempts before giving up. */ |
| IntConfOption ZOOKEEPER_CONNECTION_ATTEMPTS = |
| new IntConfOption("giraph.zkConnectionAttempts", 10, |
| "Number of ZooKeeper client connection attempts before giving up."); |
| |
| /** This directory has/stores the available checkpoint files in HDFS. */ |
| StrConfOption CHECKPOINT_DIRECTORY = |
| new StrConfOption("giraph.checkpointDirectory", "_bsp/_checkpoints/", |
| "This directory has/stores the available checkpoint files in HDFS."); |
| |
| /** |
| * Comma-separated list of directories in the local filesystem for |
| * out-of-core partitions. |
| */ |
| StrConfOption PARTITIONS_DIRECTORY = |
| new StrConfOption("giraph.partitionsDirectory", "_bsp/_partitions", |
| "Comma-separated list of directories in the local filesystem for " + |
| "out-of-core partitions."); |
| |
| /** |
| * Number of IO threads used in out-of-core mechanism. If local disk is used |
| * for spilling data to and reading data from, this number should be equal to |
| * the number of available disks on each machine. In such case, one should |
| * use giraph.partitionsDirectory to specify directories mounted on different |
| * disks. |
| */ |
| IntConfOption NUM_OUT_OF_CORE_THREADS = |
| new IntConfOption("giraph.numOutOfCoreThreads", 1, "Number of IO " + |
| "threads used in out-of-core mechanism. If using local disk to " + |
| "spill data, this should be equal to the number of available " + |
| "disks. In such case, use giraph.partitionsDirectory to specify " + |
| "mount points on different disks."); |
| |
| /** Enable out-of-core graph. */ |
| BooleanConfOption USE_OUT_OF_CORE_GRAPH = |
| new BooleanConfOption("giraph.useOutOfCoreGraph", false, |
| "Enable out-of-core graph."); |
| |
| /** Data accessor resource/object */ |
| ClassConfOption<OutOfCoreDataAccessor> OUT_OF_CORE_DATA_ACCESSOR = |
| ClassConfOption.create("giraph.outOfCoreDataAccessor", |
| LocalDiskDataAccessor.class, OutOfCoreDataAccessor.class, |
| "Data accessor used in out-of-core computation (local-disk, " + |
| "in-memory, HDFS, etc.)"); |
| |
| /** |
| * Out-of-core oracle that is to be used for adaptive out-of-core engine. If |
| * the `MAX_PARTITIONS_IN_MEMORY` is already set, this will be over-written |
| * to be `FixedPartitionsOracle`. |
| */ |
| ClassConfOption<OutOfCoreOracle> OUT_OF_CORE_ORACLE = |
| ClassConfOption.create("giraph.outOfCoreOracle", |
| MemoryEstimatorOracle.class, OutOfCoreOracle.class, |
| "Out-of-core oracle that is to be used for adaptive out-of-core " + |
| "engine"); |
| |
| /** Maximum number of partitions to hold in memory for each worker. */ |
| IntConfOption MAX_PARTITIONS_IN_MEMORY = |
| new IntConfOption("giraph.maxPartitionsInMemory", 0, |
| "Maximum number of partitions to hold in memory for each worker. By" + |
| " default it is set to 0 (for adaptive out-of-core mechanism"); |
| |
| /** Directory to write YourKit snapshots to */ |
| String YOURKIT_OUTPUT_DIR = "giraph.yourkit.outputDir"; |
| /** Default directory to write YourKit snapshots to */ |
| String YOURKIT_OUTPUT_DIR_DEFAULT = "/tmp/giraph/%JOB_ID%/%TASK_ID%"; |
| |
| /** Keep the zookeeper output for debugging? Default is to remove it. */ |
| BooleanConfOption KEEP_ZOOKEEPER_DATA = |
| new BooleanConfOption("giraph.keepZooKeeperData", false, |
| "Keep the zookeeper output for debugging? Default is to remove it."); |
| /** Default ZooKeeper snap count. */ |
| int DEFAULT_ZOOKEEPER_SNAP_COUNT = 50000; |
| /** Default ZooKeeper tick time. */ |
| int DEFAULT_ZOOKEEPER_TICK_TIME = 6000; |
| /** Default ZooKeeper maximum client connections. */ |
| int DEFAULT_ZOOKEEPER_MAX_CLIENT_CNXNS = 10000; |
| /** Number of snapshots to be retained after purge */ |
| int ZOOKEEPER_SNAP_RETAIN_COUNT = 3; |
| /** Zookeeper purge interval in hours */ |
| int ZOOKEEPER_PURGE_INTERVAL = 1; |
| /** ZooKeeper minimum session timeout */ |
| IntConfOption ZOOKEEPER_MIN_SESSION_TIMEOUT = |
| new IntConfOption("giraph.zKMinSessionTimeout", MINUTES.toMillis(10), |
| "ZooKeeper minimum session timeout"); |
| /** ZooKeeper maximum session timeout */ |
| IntConfOption ZOOKEEPER_MAX_SESSION_TIMEOUT = |
| new IntConfOption("giraph.zkMaxSessionTimeout", MINUTES.toMillis(15), |
| "ZooKeeper maximum session timeout"); |
| |
| /** ZooKeeper force sync */ |
| BooleanConfOption ZOOKEEPER_FORCE_SYNC = |
| new BooleanConfOption("giraph.zKForceSync", false, |
| "ZooKeeper force sync"); |
| |
| /** ZooKeeper skip ACLs */ |
| BooleanConfOption ZOOKEEPER_SKIP_ACL = |
| new BooleanConfOption("giraph.ZkSkipAcl", true, "ZooKeeper skip ACLs"); |
| |
| /** |
| * Whether to use SASL with DIGEST and Hadoop Job Tokens to authenticate |
| * and authorize Netty BSP Clients to Servers. |
| */ |
| BooleanConfOption AUTHENTICATE = |
| new BooleanConfOption("giraph.authenticate", false, |
| "Whether to use SASL with DIGEST and Hadoop Job Tokens to " + |
| "authenticate and authorize Netty BSP Clients to Servers."); |
| |
| /** |
| * Whether to use SSL to authenticate and authorize " |
| * Netty BSP Clients to Servers. |
| */ |
| BooleanConfOption SSL_ENCRYPT = |
| new BooleanConfOption("giraph.sslEncrypt", false, |
| "Whether to use SSL to authenticate and authorize " + |
| "Netty BSP Clients to Servers."); |
| |
| /** Use unsafe serialization? */ |
| BooleanConfOption USE_UNSAFE_SERIALIZATION = |
| new BooleanConfOption("giraph.useUnsafeSerialization", true, |
| "Use unsafe serialization?"); |
| |
| /** |
| * Use BigDataIO for messages? If there are super-vertices in the |
| * graph which receive a lot of messages (total serialized size of messages |
| * goes beyond the maximum size of a byte array), setting this option to true |
| * will remove that limit. The maximum memory available for a single vertex |
| * will be limited to the maximum heap size available. |
| */ |
| BooleanConfOption USE_BIG_DATA_IO_FOR_MESSAGES = |
| new BooleanConfOption("giraph.useBigDataIOForMessages", false, |
| "Use BigDataIO for messages?"); |
| |
| /** |
| * Maximum number of attempts a master/worker will retry before killing |
| * the job. This directly maps to the number of map task attempts in |
| * Hadoop. |
| */ |
| IntConfOption MAX_TASK_ATTEMPTS = |
| new IntConfOption("mapred.map.max.attempts", -1, |
| "Maximum number of attempts a master/worker will retry before " + |
| "killing the job. This directly maps to the number of map task " + |
| "attempts in Hadoop."); |
| |
| /** Interface to use for hostname resolution */ |
| StrConfOption DNS_INTERFACE = |
| new StrConfOption("giraph.dns.interface", "default", |
| "Interface to use for hostname resolution"); |
| /** Server for hostname resolution */ |
| StrConfOption DNS_NAMESERVER = |
| new StrConfOption("giraph.dns.nameserver", "default", |
| "Server for hostname resolution"); |
| |
| /** |
| * The application will halt after this many supersteps is completed. For |
| * instance, if it is set to 3, the application will run at most 0, 1, |
| * and 2 supersteps and then go into the shutdown superstep. |
| */ |
| IntConfOption MAX_NUMBER_OF_SUPERSTEPS = |
| new IntConfOption("giraph.maxNumberOfSupersteps", 1, |
| "The application will halt after this many supersteps is " + |
| "completed. For instance, if it is set to 3, the application will " + |
| "run at most 0, 1, and 2 supersteps and then go into the shutdown " + |
| "superstep."); |
| |
| /** |
| * The application will not mutate the graph topology (the edges). It is used |
| * to optimise out-of-core graph, by not writing back edges every time. |
| */ |
| BooleanConfOption STATIC_GRAPH = |
| new BooleanConfOption("giraph.isStaticGraph", false, |
| "The application will not mutate the graph topology (the edges). " + |
| "It is used to optimise out-of-core graph, by not writing back " + |
| "edges every time."); |
| |
| /** |
| * This option will tell which message encode & store enum to use when |
| * combining is not enabled |
| */ |
| EnumConfOption<MessageEncodeAndStoreType> MESSAGE_ENCODE_AND_STORE_TYPE = |
| EnumConfOption.create("giraph.messageEncodeAndStoreType", |
| MessageEncodeAndStoreType.class, |
| MessageEncodeAndStoreType.BYTEARRAY_PER_PARTITION, |
| "Select the message_encode_and_store_type to use"); |
| |
| /** |
| * This option can be used to specify if a source vertex present in edge |
| * input but not in vertex input can be created |
| */ |
| BooleanConfOption CREATE_EDGE_SOURCE_VERTICES = |
| new BooleanConfOption("giraph.createEdgeSourceVertices", true, |
| "Create a source vertex if present in edge input but not " + |
| "necessarily in vertex input"); |
| |
| /** |
| * Defines a call back that can be used to make decisions on |
| * whether the vertex should be created or not in the runtime. |
| */ |
| ClassConfOption<CreateSourceVertexCallback> |
| CREATE_EDGE_SOURCE_VERTICES_CALLBACK = |
| ClassConfOption.create("giraph.createEdgeSourceVerticesCallback", |
| DefaultCreateSourceVertexCallback.class, |
| CreateSourceVertexCallback.class, |
| "Decide whether we should create a source vertex when id is " + |
| "present in the edge input but not in vertex input"); |
| |
| /** |
| * This counter group will contain one counter whose name is the ZooKeeper |
| * server:port which this job is using |
| */ |
| String ZOOKEEPER_SERVER_PORT_COUNTER_GROUP = "Zookeeper server:port"; |
| |
| /** |
| * This counter group will contain one counter whose name is the ZooKeeper |
| * node path which should be created to trigger computation halt |
| */ |
| String ZOOKEEPER_HALT_NODE_COUNTER_GROUP = "Zookeeper halt node"; |
| |
| /** |
| * This counter group will contain one counter whose name is the ZooKeeper |
| * node path which contains all data about this job |
| */ |
| String ZOOKEEPER_BASE_PATH_COUNTER_GROUP = "Zookeeper base path"; |
| |
| /** |
| * Which class to use to write instructions on how to halt the application |
| */ |
| ClassConfOption<HaltApplicationUtils.HaltInstructionsWriter> |
| HALT_INSTRUCTIONS_WRITER_CLASS = ClassConfOption.create( |
| "giraph.haltInstructionsWriter", |
| HaltApplicationUtils.DefaultHaltInstructionsWriter.class, |
| HaltApplicationUtils.HaltInstructionsWriter.class, |
| "Class used to write instructions on how to halt the application"); |
| |
| /** |
| * Maximum timeout (in milliseconds) for waiting for all tasks |
| * to complete after the job is done. Defaults to 15 minutes. |
| */ |
| IntConfOption WAIT_TASK_DONE_TIMEOUT_MS = |
| new IntConfOption("giraph.waitTaskDoneTimeoutMs", MINUTES.toMillis(15), |
| "Maximum timeout (in ms) for waiting for all all tasks to " + |
| "complete"); |
| |
| /** Whether to track job progress on client or not */ |
| BooleanConfOption TRACK_JOB_PROGRESS_ON_CLIENT = |
| new BooleanConfOption("giraph.trackJobProgressOnClient", false, |
| "Whether to track job progress on client or not"); |
| |
| /** Class to use as the job progress client */ |
| ClassConfOption<JobProgressTrackerClient> JOB_PROGRESS_TRACKER_CLIENT_CLASS = |
| ClassConfOption.create("giraph.jobProgressTrackerClientClass", |
| RetryableJobProgressTrackerClient.class, |
| JobProgressTrackerClient.class, |
| "Class to use to make calls to the job progress tracker service"); |
| |
| /** Class to use to track job progress on client */ |
| ClassConfOption<JobProgressTrackerService> |
| JOB_PROGRESS_TRACKER_SERVICE_CLASS = |
| ClassConfOption.create("giraph.jobProgressTrackerServiceClass", |
| DefaultJobProgressTrackerService.class, |
| JobProgressTrackerService.class, |
| "Class to use to track job progress on client"); |
| |
| /** |
| * Minimum number of vertices to compute before adding to worker progress. |
| */ |
| LongConfOption VERTICES_TO_UPDATE_PROGRESS = |
| new LongConfOption("giraph.VerticesToUpdateProgress", 100000, |
| "Minimum number of vertices to compute before " + |
| "updating worker progress"); |
| |
| |
| /** Number of retries for creating the HDFS files */ |
| IntConfOption HDFS_FILE_CREATION_RETRIES = |
| new IntConfOption("giraph.hdfs.file.creation.retries", 10, |
| "Retries to create an HDFS file before failing"); |
| |
| /** Number of milliseconds to wait before retrying HDFS file creation */ |
| IntConfOption HDFS_FILE_CREATION_RETRY_WAIT_MS = |
| new IntConfOption("giraph.hdfs.file.creation.retry.wait.ms", 30_000, |
| "Milliseconds to wait prior to retrying creation of an HDFS file"); |
| |
| /** Number of threads for writing and reading checkpoints */ |
| IntConfOption NUM_CHECKPOINT_IO_THREADS = |
| new IntConfOption("giraph.checkpoint.io.threads", 8, |
| "Number of threads for writing and reading checkpoints"); |
| |
| /** |
| * Compression algorithm to be used for checkpointing. |
| * Defined by extension for hadoop compatibility reasons. |
| */ |
| StrConfOption CHECKPOINT_COMPRESSION_CODEC = |
| new StrConfOption("giraph.checkpoint.compression.codec", |
| ".deflate", |
| "Defines compression algorithm we will be using for " + |
| "storing checkpoint. Available options include but " + |
| "not restricted to: .deflate, .gz, .bz2, .lzo"); |
| |
| /** |
| * Defines if and when checkpointing is supported by this job. |
| * By default checkpointing is always supported unless output during the |
| * computation is enabled. |
| */ |
| ClassConfOption<CheckpointSupportedChecker> CHECKPOINT_SUPPORTED_CHECKER = |
| ClassConfOption.create("giraph.checkpoint.supported.checker", |
| DefaultCheckpointSupportedChecker.class, |
| CheckpointSupportedChecker.class, |
| "This is the way to specify if checkpointing is " + |
| "supported by the job"); |
| |
| |
| /** Number of threads to use in async message store, 0 means |
| * we should not use async message processing */ |
| IntConfOption ASYNC_MESSAGE_STORE_THREADS_COUNT = |
| new IntConfOption("giraph.async.message.store.threads", 0, |
| "Number of threads to be used in async message store."); |
| |
| /** Output format class for hadoop to use (for committing) */ |
| ClassConfOption<OutputFormat> HADOOP_OUTPUT_FORMAT_CLASS = |
| ClassConfOption.create("giraph.hadoopOutputFormatClass", |
| BspOutputFormat.class, OutputFormat.class, |
| "Output format class for hadoop to use (for committing)"); |
| |
| /** |
| * For worker to worker communication we can use IPs or host names, by |
| * default prefer IPs. |
| */ |
| BooleanConfOption PREFER_IP_ADDRESSES = |
| new BooleanConfOption("giraph.preferIP", false, |
| "Prefer IP addresses instead of host names"); |
| |
| /** |
| * Timeout for "waitForever", when we need to wait for zookeeper. |
| * Since we should never really have to wait forever. |
| * We should only wait some reasonable but large amount of time. |
| */ |
| LongConfOption WAIT_ZOOKEEPER_TIMEOUT_MSEC = |
| new LongConfOption("giraph.waitZookeeperTimeoutMsec", |
| MINUTES.toMillis(15), |
| "How long should we stay in waitForever loops in various " + |
| "places that require network connection"); |
| |
| /** |
| * Timeout for "waitForever", when we need to wait for other workers |
| * to complete their job. |
| * Since we should never really have to wait forever. |
| * We should only wait some reasonable but large amount of time. |
| */ |
| LongConfOption WAIT_FOR_OTHER_WORKERS_TIMEOUT_MSEC = |
| new LongConfOption("giraph.waitForOtherWorkersMsec", |
| HOURS.toMillis(48), |
| "How long should workers wait to finish superstep"); |
| |
| /** Number of supersteps job will run for */ |
| IntConfOption SUPERSTEP_COUNT = new IntConfOption("giraph.numSupersteps", -1, |
| "Number of supersteps job will run for"); |
| |
| /** Whether to disable GiraphClassResolver which is an efficient |
| * implementation of kryo class resolver. By default this resolver is used by |
| * KryoSimpleWritable and KryoSimpleWrapper, and can be disabled with this |
| * option */ |
| BooleanConfOption DISABLE_GIRAPH_CLASS_RESOLVER = |
| new BooleanConfOption("giraph.disableGiraphClassResolver", false, |
| "Disables GiraphClassResolver, which is a custom implementation " + |
| "of kryo class resolver that avoids writing class names to the " + |
| "underlying stream for faster serialization."); |
| |
| /** |
| * Path where jmap exists |
| */ |
| StrConfOption JMAP_PATH = new StrConfOption("giraph.jmapPath", "jmap", |
| "Path to use for invoking jmap"); |
| |
| /** |
| * Whether to fail the job or just warn when input is empty |
| */ |
| BooleanConfOption FAIL_ON_EMPTY_INPUT = new BooleanConfOption( |
| "giraph.failOnEmptyInput", true, |
| "Whether to fail the job or just warn when input is empty"); |
| |
| /** SSLEventHandler class - optional */ |
| ClassConfOption<SSLEventHandler> SSL_EVENT_HANDLER_CLASS = |
| ClassConfOption.create("giraph.sslEventHandler", |
| null, SSLEventHandler.class, |
| "SSLEventHandler class - optional"); |
| } |
| // CHECKSTYLE: resume InterfaceIsTypeCheck |
