accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bil...@apache.org
Subject svn commit: r1229248 - in /incubator/accumulo/branches/1.4/src/core/src: main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
Date Mon, 09 Jan 2012 16:49:56 GMT
Author: billie
Date: Mon Jan  9 16:49:56 2012
New Revision: 1229248

URL: http://svn.apache.org/viewvc?rev=1229248&view=rev
Log:
ACCUMULO-287 added mock instance to AccumuloOutputFormat

Added:
    incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
  (with props)
Modified:
    incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java

Modified: incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java?rev=1229248&r1=1229247&r2=1229248&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java
(original)
+++ incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormat.java
Mon Jan  9 16:49:56 2012
@@ -30,6 +30,7 @@ import org.apache.accumulo.core.client.M
 import org.apache.accumulo.core.client.TableExistsException;
 import org.apache.accumulo.core.client.TableNotFoundException;
 import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.mock.MockInstance;
 import org.apache.accumulo.core.data.ColumnUpdate;
 import org.apache.accumulo.core.data.KeyExtent;
 import org.apache.accumulo.core.data.Mutation;
@@ -72,6 +73,7 @@ public class AccumuloOutputFormat extend
   
   private static final String INSTANCE_NAME = PREFIX + ".instanceName";
   private static final String ZOOKEEPERS = PREFIX + ".zooKeepers";
+  private static final String MOCK = ".useMockInstance";
   
   private static final String CREATETABLES = PREFIX + ".createtables";
   private static final String LOGLEVEL = PREFIX + ".loglevel";
@@ -125,6 +127,13 @@ public class AccumuloOutputFormat extend
     conf.set(ZOOKEEPERS, zooKeepers);
   }
   
+  public static void setMockInstance(JobContext job, String instanceName) {
+    Configuration conf = job.getConfiguration();
+    conf.setBoolean(INSTANCE_HAS_BEEN_SET, true);
+    conf.setBoolean(MOCK, true);
+    conf.set(INSTANCE_NAME, instanceName);
+  }
+  
   public static void setMaxMutationBufferSize(JobContext job, long numberOfBytes) {
     job.getConfiguration().setLong(MAX_MUTATION_BUFFER_SIZE, numberOfBytes);
   }
@@ -168,6 +177,8 @@ public class AccumuloOutputFormat extend
   
   protected static Instance getInstance(JobContext job) {
     Configuration conf = job.getConfiguration();
+    if (conf.getBoolean(MOCK, false))
+      return new MockInstance(conf.get(INSTANCE_NAME));
     return new ZooKeeperInstance(conf.get(INSTANCE_NAME), conf.get(ZOOKEEPERS));
   }
   

Added: incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java?rev=1229248&view=auto
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
(added)
+++ incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
Mon Jan  9 16:49:56 2012
@@ -0,0 +1,126 @@
+/**
+ * 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.accumulo.core.client.mapreduce;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map.Entry;
+
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.mock.MockInstance;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.junit.Test;
+
+/**
+ * 
+ */
+public class AccumuloOutputFormatTest {
+  static class TestMapper extends Mapper<Key,Value,Text,Mutation> {
+    Key key = null;
+    int count = 0;
+    
+    @Override
+    protected void map(Key k, Value v, Context context) throws IOException, InterruptedException
{
+      if (key != null)
+        assertEquals(key.getRow().toString(), new String(v.get()));
+      assertEquals(k.getRow(), new Text(String.format("%09x", count + 1)));
+      assertEquals(new String(v.get()), String.format("%09x", count));
+      key = new Key(k);
+      count++;
+    }
+    
+    @Override
+    protected void cleanup(Context context) throws IOException, InterruptedException {
+      super.cleanup(context);
+      Mutation m = new Mutation("total");
+      m.put("", "", Integer.toString(count));
+      try {
+        context.write(new Text(), m);
+      } catch (NullPointerException e) {}
+    }
+  }
+  
+  @Test
+  public void testMR() throws Exception {
+    MockInstance mockInstance = new MockInstance("testmrinstance");
+    Connector c = mockInstance.getConnector("root", new byte[] {});
+    c.tableOperations().create("testtable1");
+    c.tableOperations().create("testtable2");
+    BatchWriter bw = c.createBatchWriter("testtable1", 10000L, 1000L, 4);
+    for (int i = 0; i < 100; i++) {
+      Mutation m = new Mutation(new Text(String.format("%09x", i + 1)));
+      m.put(new Text(), new Text(), new Value(String.format("%09x", i).getBytes()));
+      bw.addMutation(m);
+    }
+    bw.close();
+    
+    Job job = new Job();
+    job.setInputFormatClass(AccumuloInputFormat.class);
+    job.setMapperClass(TestMapper.class);
+    job.setOutputFormatClass(AccumuloOutputFormat.class);
+    job.setOutputKeyClass(Text.class);
+    job.setOutputValueClass(Mutation.class);
+    job.setNumReduceTasks(0);
+    AccumuloInputFormat.setInputInfo(job.getConfiguration(), "root", "".getBytes(), "testtable1",
new Authorizations());
+    AccumuloInputFormat.setMockInstance(job.getConfiguration(), "testmrinstance");
+    AccumuloOutputFormat.setOutputInfo(job, "root", "".getBytes(), false, "testtable2");
+    AccumuloOutputFormat.setMockInstance(job, "testmrinstance");
+    
+    AccumuloInputFormat input = new AccumuloInputFormat();
+    List<InputSplit> splits = input.getSplits(job);
+    assertEquals(splits.size(), 1);
+    
+    AccumuloOutputFormat output = new AccumuloOutputFormat();
+    
+    TestMapper mapper = (TestMapper) job.getMapperClass().newInstance();
+    for (InputSplit split : splits) {
+      TaskAttemptID id = new TaskAttemptID();
+      TaskAttemptContext tac = new TaskAttemptContext(job.getConfiguration(), id);
+      RecordReader<Key,Value> reader = input.createRecordReader(split, tac);
+      RecordWriter<Text,Mutation> writer = output.getRecordWriter(tac);
+      Mapper<Key,Value,Text,Mutation>.Context context = mapper.new Context(job.getConfiguration(),
id, reader, writer, null, null, split);
+      reader.initialize(split, context);
+      mapper.run(context);
+      writer.close(context);
+    }
+    
+    Scanner scanner = c.createScanner("testtable2", new Authorizations());
+    Iterator<Entry<Key,Value>> iter = scanner.iterator();
+    assertTrue(iter.hasNext());
+    Entry<Key,Value> entry = iter.next();
+    assertEquals(Integer.parseInt(new String(entry.getValue().get())), 100);
+    assertFalse(iter.hasNext());
+  }
+}

Propchange: incubator/accumulo/branches/1.4/src/core/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
------------------------------------------------------------------------------
    svn:eol-style = native



Mime
View raw message