giraph-accumulo/src/main/java/org/apache/giraph/io/accumulo/AccumuloVertexOutputFormat.java

/*
 * 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.io.accumulo;

import java.io.IOException;
import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
import org.apache.accumulo.core.data.Mutation;
import org.apache.giraph.io.VertexOutputFormat;
import org.apache.giraph.io.VertexWriter;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
/**
 *
 *  Class which wraps the AccumuloOutputFormat. It's designed
 *  as an extension point to VertexOutputFormat subclasses who wish
 *  to write vertices back to an Accumulo table.
 *
 *  Works with
 *  {@link AccumuloVertexInputFormat}
 *
 *
 * @param <I> vertex id type
 * @param <V>  vertex value type
 * @param <E>  edge type
 */
public abstract class AccumuloVertexOutputFormat<
        I extends WritableComparable,
        V extends Writable,
        E extends Writable>
        extends VertexOutputFormat<I, V, E> {


  /**
   * Output table parameter
   */
  public static final String OUTPUT_TABLE = "OUTPUT_TABLE";

  /**
   * Accumulo delegate for table output
   */
  protected AccumuloOutputFormat accumuloOutputFormat =
          new AccumuloOutputFormat();

  /**
   *
   * Main abstraction point for vertex writers to persist back
   * to Accumulo tables.
   *
   * @param <I> vertex id type
   * @param <V> vertex value type
   * @param <E>  edge type
   */
  public abstract static class AccumuloVertexWriter<
      I extends WritableComparable,
      V extends Writable,
      E extends Writable>
      extends VertexWriter<I, V, E> {

    /**
     * task attempt context.
     */
    private TaskAttemptContext context;

    /**
     * Accumulo record writer
     */
    private RecordWriter<Text, Mutation> recordWriter;

    /**
     * Constructor for use with subclasses
     *
     * @param recordWriter accumulo record writer
     */
    public AccumuloVertexWriter(RecordWriter<Text, Mutation> recordWriter) {
      this.recordWriter = recordWriter;
    }

    /**
     * initialize
     *
     * @param context Context used to write the vertices.
     * @throws IOException
     */
    public void initialize(TaskAttemptContext context) throws IOException {
      this.context = context;
    }

    /**
     *  close
     *
     * @param context the context of the task
     * @throws IOException
     * @throws InterruptedException
     */
    public void close(TaskAttemptContext context)
      throws IOException, InterruptedException {
      recordWriter.close(context);
    }

    /**
     * Get the table record writer;
     *
     * @return Record writer to be used for writing.
     */
    public RecordWriter<Text, Mutation> getRecordWriter() {
      return recordWriter;
    }

    /**
     * Get the context.
     *
     * @return Context passed to initialize.
     */
    public TaskAttemptContext getContext() {
      return context;
    }

  }
  /**
   *
   * checkOutputSpecs
   *
   * @param context information about the job
   * @throws IOException
   * @throws InterruptedException
   */
  @Override
  public void checkOutputSpecs(JobContext context)
    throws IOException, InterruptedException {
    try {
      accumuloOutputFormat.checkOutputSpecs(context);
    } catch (IOException e) {
      if (e.getMessage().contains("Output info has not been set")) {
        throw new IOException(e.getMessage() + " Make sure you initialized" +
                " AccumuloOutputFormat static setters " +
                "before passing the config to GiraphJob.");
      }
    }
  }

  /**
   * getOutputCommitter
   *
   * @param context the task context
   * @return OutputCommitter
   * @throws IOException
   * @throws InterruptedException
   */
  @Override
  public OutputCommitter getOutputCommitter(TaskAttemptContext context)
    throws IOException, InterruptedException {
    return accumuloOutputFormat.getOutputCommitter(context);
  }
}