Accumulo tables can be used as the source and destination of MapReduce jobs.

General MapReduce configuration

Add Accumulo’s MapReduce API to your dependencies

If you are using Maven, add the following dependency to your pom.xml to use Accumulo’s MapReduce API:

  1. <dependency>
  2.   <groupId>org.apache.accumulo</groupId>
  3.   <artifactId>accumulo-hadoop-mapreduce</artifactId>
  4.   <version>2.1.2</version>
  5. </dependency>

The MapReduce API consists of the following classes:

Before 2.0, the MapReduce API resided in the org.apache.accumulo.core.client package of the accumulo-core jar. While this old API still exists and can be used, it has been deprecated and will be removed eventually.

Configure dependencies for your MapReduce job

Before 2.0, Accumulo used the same versions for dependencies (such as Guava) as Hadoop. This allowed MapReduce jobs to run with both Accumulo’s & Hadoop’s dependencies on the classpath.

Since 2.0, Accumulo no longer has the same versions for dependencies as Hadoop. While this allows Accumulo to update its dependencies more frequently, it can cause problems if both Accumulo’s & Hadoop’s dependencies are on the classpath of the MapReduce job. When launching a MapReduce job that uses Accumulo, you should build a shaded jar with all of your dependencies and complete the following steps so YARN only includes Hadoop code (and not all of Hadoop’s dependencies) when running your MapReduce job:

  1. Set export HADOOP_USE_CLIENT_CLASSLOADER=true in your environment before submitting your job with yarn command.

  2. Set the following in your Job configuration.

    1.  job.getConfiguration().set("mapreduce.job.classloader", "true");

Read input from an Accumulo table

Follow the steps below to create a MapReduce job that reads from an Accumulo table:

  1. Create a Mapper with the following class parameterization.

    1.  class MyMapper extends Mapper<Key,Value,WritableComparable,Writable> {
    2.      public void map(Key k, Value v, Context c) {
    3.          // transform key and value data here
    4.      }
    5.  }

  2. Configure your MapReduce job to use AccumuloInputFormat.

    1.  Job job = Job.getInstance();
    2.  job.setInputFormatClass(AccumuloInputFormat.class);
    3.  Properties props = Accumulo.newClientProperties().to("myinstance","zoo1,zoo2")
    4.                          .as("user", "passwd").build();
    5.  AccumuloInputFormat.configure().clientProperties(props).table(table).store(job);

    AccumuloInputFormat has optional settings.

    1.  List<Range> ranges = new ArrayList<Range>();
    2.  Collection<IteratorSetting.Column> columns = new ArrayList<IteratorSetting.Column>();
    3.  // populate ranges & columns
    4.  IteratorSetting is = new IteratorSetting(30, RexExFilter.class);
    5.  RegExFilter.setRegexs(is, ".*suffix", null, null, null, true);
    7.  AccumuloInputFormat.configure().clientProperties(props).table(table)
    8.      .auths(Authorizations.EMPTY) // optional: default to user's auths if not set
    9.      .ranges(ranges)              // optional: only read specified ranges
    10.      .fetchColumns(columns)       // optional: only read specified columns
    11.      .addIterator(is)             // optional: add iterator that matches row IDs
    12.      .store(job);

    AccumuloInputFormat can also be configured to read from multiple Accumulo tables.

    1.  Job job = Job.getInstance();
    2.  job.setInputFormatClass(AccumuloInputFormat.class);
    3.  Properties props = Accumulo.newClientProperties().to("myinstance","zoo1,zoo2")
    4.                          .as("user", "passwd").build();
    5.  AccumuloInputFormat.configure().clientProperties(props)
    6.      .table("table1").auths(Authorizations.EMPTY).ranges(tableOneRanges)
    7.      .table("table2").auths(Authorizations.EMPTY).ranges(tableTwoRanges)
    8.      .store(job);

    If reading from multiple tables, the table name can be retrieved from the input split:

    1.  class MyMapper extends Mapper<Key,Value,WritableComparable,Writable> {
    2.      public void map(Key k, Value v, Context c) {
    3.          RangeInputSplit split = (RangeInputSplit)c.getInputSplit();
    4.          String tableName = split.getTableName();
    5.          // do something with table name
    6.      }
    7.  }

Write output to an Accumulo table

Follow the steps below to write to an Accumulo table from a MapReduce job.

  1. Create a Reducer with the following class parameterization. The key emitted from the Reducer identifies the table to which the mutation is sent. This allows a single Reducer to write to more than one table if desired. A default table can be configured using the AccumuloOutputFormat, in which case the output table name does not have to be passed to the Context object within the Reducer.

    1.  class MyReducer extends Reducer<WritableComparable, Writable, Text, Mutation> {
    2.      public void reduce(WritableComparable key, Iterable<Text> values, Context c) {
    3.          Mutation m;
    4.          // create the mutation based on input key and value
    5.          c.write(new Text("output-table"), m);
    6.      }
    7.  }

    The Text object passed as the output should contain the name of the table to which this mutation should be applied. The Text can be null in which case the mutation will be applied to the default table name specified in the AccumuloOutputFormat options.

  2. Configure your MapReduce job to use AccumuloOutputFormat.

    1.  Job job = Job.getInstance();
    2.  job.setOutputFormatClass(AccumuloOutputFormat.class);
    3.  Properties props = Accumulo.newClientProperties().to("myinstance","zoo1,zoo2")
    4.                          .as("user", "passwd").build();
    5.  AccumuloOutputFormat.configure().clientProperties(props)
    6.      .defaultTable("mytable").store(job);

Write output to RFiles in HDFS

Follow the steps below to have a MapReduce job output to RFiles in HDFS. These files can then be bulk imported into Accumulo:

  1. Create a Mapper or Reducer with Key & Value as output parameters.

    1.  class MyReducer extends Reducer<WritableComparable, Writable, Key, Value> {
    2.      public void reduce(WritableComparable key, Iterable<Text> values, Context c) {
    3.          Key key;
    4.          Value value;
    5.          // create Key & Value based on input
    6.          c.write(key, value);
    7.      }
    8.  }

  2. Configure your MapReduce job to use AccumuloFileOutputFormat.

    1.  Job job = Job.getInstance();
    2.  job.setOutputFormatClass(AccumuloFileOutputFormat.class);
    3.  AccumuloFileOutputFormat.configure()
    4.      .outputPath(new Path("hdfs://localhost:8020/myoutput/")).store(job);

Example Code

The Accumulo Examples repo has several MapReduce examples:

  • wordcount - Uses MapReduce and Accumulo to do a word count on text files
  • regex - Uses MapReduce and Accumulo to find data using regular expressions
  • rowhash - Uses MapReduce to read a table and write to a new column in the same table
  • tabletofile - Uses MapReduce to read a table and write one of its columns to a file in HDFS
  • uniquecols - Uses MapReduce to count unique columns in Accumulo