accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ctubb...@apache.org
Subject svn commit: r1433751 - in /accumulo/contrib/bsp/trunk: ./ src/main/java/org/apache/accumulo/bsp/ src/test/java/org/apache/accumulo/core/client/mapreduce/
Date Tue, 15 Jan 2013 23:58:38 GMT
Author: ctubbsii
Date: Tue Jan 15 23:58:38 2013
New Revision: 1433751

URL: http://svn.apache.org/viewvc?rev=1433751&view=rev
Log:
ACCUMULO-532 Update contrib to reflect changes in ACCUMULO-769 and to use the released versions
of Hama BSP so we can close this ticket

Added:
    accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/MapreduceWrapper.java
    accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatIT.java
    accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatIT.java
      - copied, changed from r1431766, accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
Removed:
    accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
Modified:
    accumulo/contrib/bsp/trunk/pom.xml
    accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloInputFormat.java
    accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloOutputFormat.java
    accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java

Modified: accumulo/contrib/bsp/trunk/pom.xml
URL: http://svn.apache.org/viewvc/accumulo/contrib/bsp/trunk/pom.xml?rev=1433751&r1=1433750&r2=1433751&view=diff
==============================================================================
--- accumulo/contrib/bsp/trunk/pom.xml (original)
+++ accumulo/contrib/bsp/trunk/pom.xml Tue Jan 15 23:58:38 2013
@@ -19,12 +19,113 @@
   <groupId>org.apache.accumulo</groupId>
   <artifactId>accumulo-bsp</artifactId>
   <version>1.5.0-SNAPSHOT</version>
-  
+
+  <parent>
+    <groupId>org.apache</groupId>
+    <artifactId>apache</artifactId>
+    <version>12</version>
+  </parent>
+
+  <properties>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+  </properties>
+
+  <build>
+    <pluginManagement>
+      <plugins>
+        <plugin>
+          <groupId>org.eclipse.m2e</groupId>
+          <artifactId>lifecycle-mapping</artifactId>
+          <version>1.0.0</version>
+          <configuration>
+            <lifecycleMappingMetadata>
+              <pluginExecutions>
+                <pluginExecution>
+                  <pluginExecutionFilter>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-remote-resources-plugin</artifactId>
+                    <versionRange>[1.0,)</versionRange>
+                    <goals>
+                      <goal>process</goal>
+                    </goals>
+                  </pluginExecutionFilter>
+                  <action>
+                    <ignore />
+                  </action>
+                </pluginExecution>
+              </pluginExecutions>
+            </lifecycleMappingMetadata>
+          </configuration>
+        </plugin>
+        <plugin>
+          <groupId>org.apache.maven.plugins</groupId>
+          <artifactId>maven-compiler-plugin</artifactId>
+          <configuration>
+            <source>1.6</source>
+            <target>1.6</target>
+          </configuration>
+        </plugin>
+      </plugins>
+    </pluginManagement>
+  </build>
+
+  <profiles>
+    <profile>
+      <id>integration-tests</id>
+      <activation>
+        <property>
+          <name>!skipTests</name>
+        </property>
+      </activation>
+      <build>
+        <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-jar-plugin</artifactId>
+            <executions>
+              <execution>
+                <id>create-integration-test-jar</id>
+                <phase>pre-integration-test</phase>
+                <goals>
+                  <goal>test-jar</goal>
+                </goals>
+                <configuration>
+                  <finalName>integration</finalName>
+                </configuration>
+              </execution>
+            </executions>
+          </plugin>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-failsafe-plugin</artifactId>
+            <executions>
+              <execution>
+                <id>run-integration-tests</id>
+                <phase>integration-test</phase>
+                <goals>
+                  <goal>integration-test</goal>
+                </goals>
+              </execution>
+              <execution>
+                <id>verify-integration-tests</id>
+                <phase>verify</phase>
+                <goals>
+                  <goal>verify</goal>
+                </goals>
+              </execution>
+            </executions>
+          </plugin>
+        </plugins>
+      </build>
+    </profile>
+  </profiles>
+
   <dependencies>
     <dependency>
       <groupId>org.apache.hama</groupId>
       <artifactId>hama-core</artifactId>
-      <version>0.4.0-incubating</version>
+      <version>0.6.0</version>
     </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
@@ -34,7 +135,13 @@
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-core</artifactId>
-      <version>0.20.2</version>
+      <version>[1.0.0,2.0.0)</version>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>4.11</version>
+      <scope>test</scope>
     </dependency>
   </dependencies>
 </project>

Modified: accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloInputFormat.java
URL: http://svn.apache.org/viewvc/accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloInputFormat.java?rev=1433751&r1=1433750&r2=1433751&view=diff
==============================================================================
--- accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloInputFormat.java
(original)
+++ accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloInputFormat.java
Tue Jan 15 23:58:38 2013
@@ -27,23 +27,25 @@ import org.apache.hama.bsp.InputFormat;
 import org.apache.hama.bsp.InputSplit;
 import org.apache.hama.bsp.RecordReader;
 
+/**
+ * <p>
+ * AccumuloInputFormat class. To be used with Hama BSP.
+ * </p>
+ * 
+ * @see BSPJob#setInputFormat(Class)
+ */
 public class AccumuloInputFormat extends org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat
implements InputFormat<Key,Value> {
+  
   public class BSPRecordReaderBase extends RecordReaderBase<Key,Value> implements RecordReader<Key,Value>
{
     public BSPRecordReaderBase(InputSplit split, BSPJob job) throws IOException {
-      this.initialize((BSPRangeInputSplit) split, job.getConf());
+      this.initialize((BSPRangeInputSplit) split, MapreduceWrapper.wrappedTaskAttemptContext(job));
     }
     
-    /*
-     * @see org.apache.hadoop.mapreduce.RecordReader#nextKeyValue()
-     */
     @Override
     public boolean nextKeyValue() throws IOException, InterruptedException {
       return next(currentKey, currentValue);
     }
     
-    /*
-     * @see org.apache.hama.bsp.RecordReader#createKey()
-     */
     @Override
     public Key createKey() {
       if (currentKey == null) {
@@ -53,9 +55,6 @@ public class AccumuloInputFormat extends
       }
     }
     
-    /*
-     * @see org.apache.hama.bsp.RecordReader#createValue()
-     */
     @Override
     public Value createValue() {
       if (currentValue == null) {
@@ -65,17 +64,11 @@ public class AccumuloInputFormat extends
       }
     }
     
-    /*
-     * @see org.apache.hama.bsp.RecordReader#getPos()
-     */
     @Override
     public long getPos() throws IOException {
       return 0;
     }
     
-    /*
-     * @see org.apache.hama.bsp.RecordReader#next(java.lang.Object, java.lang.Object)
-     */
     @Override
     public boolean next(Key k, Value v) throws IOException {
       if (scannerIterator.hasNext()) {
@@ -108,7 +101,7 @@ public class AccumuloInputFormat extends
   
   @Override
   public InputSplit[] getSplits(BSPJob job, int arg1) throws IOException {
-    List<org.apache.hadoop.mapreduce.InputSplit> splits = getSplits(job.getConf());
+    List<org.apache.hadoop.mapreduce.InputSplit> splits = getSplits(MapreduceWrapper.wrappedTaskAttemptContext(job));
     InputSplit[] bspSplits = new BSPRangeInputSplit[splits.size()];
     for (int i = 0; i < splits.size(); i++) {
       bspSplits[i] = new BSPRangeInputSplit((RangeInputSplit) splits.get(i));

Modified: accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloOutputFormat.java
URL: http://svn.apache.org/viewvc/accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloOutputFormat.java?rev=1433751&r1=1433750&r2=1433751&view=diff
==============================================================================
--- accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloOutputFormat.java
(original)
+++ accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/AccumuloOutputFormat.java
Tue Jan 15 23:58:38 2013
@@ -21,27 +21,34 @@ import java.io.IOException;
 import org.apache.accumulo.core.client.AccumuloException;
 import org.apache.accumulo.core.client.AccumuloSecurityException;
 import org.apache.accumulo.core.data.Mutation;
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.io.Text;
 import org.apache.hama.bsp.BSPJob;
 import org.apache.hama.bsp.OutputFormat;
 import org.apache.hama.bsp.RecordWriter;
 
+/**
+ * <p>
+ * AccumuloOutputFormat class. To be used with Hama BSP.
+ * </p>
+ * 
+ * @see BSPJob#setOutputFormat(Class)
+ */
 public class AccumuloOutputFormat extends org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat
implements OutputFormat<Text,Mutation> {
   
   protected static class BSPRecordWriter extends AccumuloRecordWriter implements RecordWriter<Text,Mutation>
{
-    BSPRecordWriter(Configuration conf) throws AccumuloException, AccumuloSecurityException,
IOException {
-      super(conf);
+    
+    private BSPJob job;
+    
+    BSPRecordWriter(BSPJob job) throws AccumuloException, AccumuloSecurityException, IOException
{
+      super(MapreduceWrapper.wrappedTaskAttemptContext(job));
+      this.job = job;
     }
     
-    /*
-     * @see org.apache.hama.bsp.RecordWriter#close()
-     */
     @Override
     public void close() throws IOException {
       try {
-        close(null);
+        close(MapreduceWrapper.wrappedTaskAttemptContext(job));
       } catch (InterruptedException e) {
         throw new IOException(e);
       }
@@ -49,21 +56,15 @@ public class AccumuloOutputFormat extend
     
   }
   
-  /*
-   * @see org.apache.hama.bsp.OutputFormat#checkOutputSpecs(org.apache.hadoop.fs.FileSystem,
org.apache.hama.bsp.BSPJob)
-   */
   @Override
   public void checkOutputSpecs(FileSystem fs, BSPJob job) throws IOException {
-    checkOutputSpecs(job.getConf());
+    checkOutputSpecs(MapreduceWrapper.wrappedTaskAttemptContext(job));
   }
   
-  /*
-   * @see org.apache.hama.bsp.OutputFormat#getRecordWriter(org.apache.hadoop.fs.FileSystem,
org.apache.hama.bsp.BSPJob, java.lang.String)
-   */
   @Override
-  public RecordWriter<Text,Mutation> getRecordWriter(FileSystem fs, BSPJob job, String
arg2) throws IOException {
+  public RecordWriter<Text,Mutation> getRecordWriter(FileSystem fs, BSPJob job, String
name) throws IOException {
     try {
-      return new BSPRecordWriter(job.getConf());
+      return new BSPRecordWriter(job);
     } catch (Exception e) {
       throw new IOException(e);
     }

Added: accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/MapreduceWrapper.java
URL: http://svn.apache.org/viewvc/accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/MapreduceWrapper.java?rev=1433751&view=auto
==============================================================================
--- accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/MapreduceWrapper.java
(added)
+++ accumulo/contrib/bsp/trunk/src/main/java/org/apache/accumulo/bsp/MapreduceWrapper.java
Tue Jan 15 23:58:38 2013
@@ -0,0 +1,69 @@
+/**
+ * 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.bsp;
+
+import java.io.IOException;
+
+import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
+import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hama.bsp.BSPJob;
+
+/**
+ * <p>
+ * MapreduceWrapper class. Provides a wrapper to wrap {@link BSPJob} into the appropriate
Hadoop type required by {@link AccumuloInputFormat} and
+ * {@link AccumuloOutputFormat} static configurator methods. Useful for reusing code to set
the job's configuration and not using the expected Hadoop API.
+ * </p>
+ */
+public class MapreduceWrapper {
+  
+  /**
+   * Wraps a {@link BSPJob} for reading its {@link Configuration} within Accumulo MapReduce
classes' protected static configuration getters.
+   * 
+   * @param job
+   *          the {@link BSPJob} instance to be wrapped
+   * @return an instance of {@link TaskAttemptContext} whose {@link Configuration} is the
same as the job
+   */
+  public static TaskAttemptContext wrappedTaskAttemptContext(final BSPJob job) {
+    return new TaskAttemptContext(job.getConfiguration(), new TaskAttemptID());
+  }
+  
+  /**
+   * Wraps a {@link BSPJob} for writing its {@link Configuration} within Accumulo MapReduce
classes' public static configuration setters.
+   * 
+   * @param job
+   *          the {@link BSPJob} instance to be wrapped
+   * @return an instance of {@link Job} that exposes {@link BSPJob#getConfiguration()} via
{@link Job#getConfiguration()}; no other methods of {@link Job} are
+   *         implemented, so this object cannot be used for anything other than editing the
{@link BSPJob}'s {@link Configuration}
+   */
+  public static Job wrappedJob(BSPJob job) {
+    final BSPJob bspJob = job;
+    try {
+      return new Job() {
+        @Override
+        public Configuration getConfiguration() {
+          return bspJob.getConfiguration();
+        }
+      };
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

Added: accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatIT.java
URL: http://svn.apache.org/viewvc/accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatIT.java?rev=1433751&view=auto
==============================================================================
--- accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatIT.java
(added)
+++ accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatIT.java
Tue Jan 15 23:58:38 2013
@@ -0,0 +1,113 @@
+/**
+ * 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.fail;
+
+import java.io.IOException;
+
+import org.apache.accumulo.bsp.AccumuloInputFormat;
+import org.apache.accumulo.bsp.MapreduceWrapper;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+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.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hama.bsp.BSP;
+import org.apache.hama.bsp.BSPJob;
+import org.apache.hama.bsp.BSPPeer;
+import org.apache.hama.bsp.InputSplit;
+import org.apache.hama.bsp.sync.SyncException;
+import org.apache.hama.util.KeyValuePair;
+import org.junit.Test;
+
+/**
+ * 
+ */
+public class AccumuloInputFormatIT {
+  
+  static class InputFormatTestBSP<M extends Writable> extends BSP<Key,Value,Key,Value,M>
{
+    Key key = null;
+    int count = 0;
+    
+    @Override
+    public void bsp(BSPPeer<Key,Value,Key,Value,M> peer) throws IOException, SyncException,
InterruptedException {
+      // this method reads the next key value record from file
+      KeyValuePair<Key,Value> pair;
+      
+      while ((pair = peer.readNext()) != null) {
+        if (key != null) {
+          assertEquals(key.getRow().toString(), new String(pair.getValue().get()));
+        }
+        
+        assertEquals(pair.getKey().getRow(), new Text(String.format("%09x", count + 1)));
+        assertEquals(new String(pair.getValue().get()), String.format("%09x", count));
+        count++;
+        
+        key = new Key(pair.getKey());
+      }
+      
+      peer.sync();
+      assertEquals(100, count);
+    }
+  }
+  
+  @Test
+  public void testBSPInputFormat() throws Exception {
+    MockInstance mockInstance = new MockInstance("testmapinstance");
+    Connector c = mockInstance.getConnector("root", new byte[] {});
+    if (c.tableOperations().exists("testtable"))
+      c.tableOperations().delete("testtable");
+    c.tableOperations().create("testtable");
+    
+    BatchWriter bw = c.createBatchWriter("testtable", new BatchWriterConfig());
+    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();
+    
+    BSPJob bspJob = new BSPJob();
+    Job job = MapreduceWrapper.wrappedJob(bspJob);
+    
+    bspJob.setInputFormat(AccumuloInputFormat.class);
+    bspJob.setBspClass(InputFormatTestBSP.class);
+    bspJob.setInputPath(new Path("test"));
+    
+    AccumuloInputFormat.setInputInfo(job, "root", "".getBytes(), "testtable", new Authorizations());
+    AccumuloInputFormat.setMockInstance(job, "testmapinstance");
+    
+    AccumuloInputFormat input = new AccumuloInputFormat();
+    InputSplit[] splits = input.getSplits(bspJob, 0);
+    assertEquals(splits.length, 1);
+    
+    bspJob.setJar("target/integration-tests.jar");
+    bspJob.setOutputPath(new Path("target/bsp-inputformat-test"));
+    if (!bspJob.waitForCompletion(false))
+      fail("Job not finished successfully");
+  }
+  
+}

Modified: accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
URL: http://svn.apache.org/viewvc/accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java?rev=1433751&r1=1433750&r2=1433751&view=diff
==============================================================================
--- accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
(original)
+++ accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputFormatTest.java
Tue Jan 15 23:58:38 2013
@@ -21,243 +21,175 @@ import static org.junit.Assert.assertTru
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 
 import org.apache.accumulo.bsp.AccumuloInputFormat;
-import org.apache.accumulo.core.client.BatchWriter;
-import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.bsp.MapreduceWrapper;
 import org.apache.accumulo.core.client.IteratorSetting;
-import org.apache.accumulo.core.client.mapreduce.InputFormatBase.AccumuloIterator;
-import org.apache.accumulo.core.client.mapreduce.InputFormatBase.AccumuloIteratorOption;
-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.iterators.user.RegExFilter;
 import org.apache.accumulo.core.iterators.user.WholeRowIterator;
-import org.apache.accumulo.core.security.Authorizations;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.Text;
-import org.apache.hama.HamaConfiguration;
-import org.apache.hama.bsp.BSP;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hama.bsp.BSPJob;
-import org.apache.hama.bsp.BSPPeer;
-import org.apache.hama.bsp.InputSplit;
-import org.apache.hama.bsp.sync.SyncException;
-import org.apache.hama.util.KeyValuePair;
-import org.junit.After;
 import org.junit.Test;
 
+/**
+ * 
+ */
 public class AccumuloInputFormatTest {
   
-  @After
-  public void tearDown() throws Exception {}
-  
-  /**
-   * Test basic setting & getting of max versions.
-   * 
-   * @throws IOException
-   *           Signals that an I/O exception has occurred.
-   */
-  @Test
-  public void testMaxVersions() throws IOException {
-    BSPJob job = new BSPJob();
-    AccumuloInputFormat.setMaxVersions(job.getConf(), 1);
-    int version = AccumuloInputFormat.getMaxVersions(job.getConf());
-    assertEquals(1, version);
-  }
-  
-  @Test(expected = IOException.class)
-  public void testMaxVersionsLessThan1() throws IOException {
-    BSPJob job = new BSPJob();
-    AccumuloInputFormat.setMaxVersions(job.getConf(), 0);
-  }
-  
-  @Test
-  public void testNoMaxVersion() throws IOException {
-    BSPJob job = new BSPJob();
-    assertEquals(-1, AccumuloInputFormat.getMaxVersions(job.getConf()));
-  }
-  
   @Test
   public void testSetIterator() throws IOException {
-    BSPJob job = new BSPJob();
+    BSPJob bspJob = new BSPJob();
     
-    AccumuloInputFormat.addIterator(job.getConf(), new IteratorSetting(1, "WholeRow", "org.apache.accumulo.core.iterators.WholeRowIterator"));
-    Configuration conf = job.getConf();
-    String iterators = conf.get("AccumuloInputFormat.iterators");
-    assertEquals("1:org.apache.accumulo.core.iterators.WholeRowIterator:WholeRow", iterators);
+    Job job = MapreduceWrapper.wrappedJob(bspJob);
+    AccumuloInputFormat.addIterator(job, new IteratorSetting(1, "WholeRow", "org.apache.accumulo.core.iterators.WholeRowIterator"));
+    
+    TaskAttemptContext context = MapreduceWrapper.wrappedTaskAttemptContext(bspJob);
+    List<IteratorSetting> iterators = AccumuloInputFormat.getIterators(context);
+    assertEquals(1, iterators.size());
+    IteratorSetting iter = iterators.get(0);
+    assertEquals(1, iter.getPriority());
+    assertEquals("org.apache.accumulo.core.iterators.WholeRowIterator", iter.getIteratorClass());
+    assertEquals("WholeRow", iter.getName());
+    assertEquals(0, iter.getOptions().size());
   }
   
   @Test
   public void testAddIterator() throws IOException {
-    BSPJob job = new BSPJob();
+    BSPJob bspJob = new BSPJob();
     
-    AccumuloInputFormat.addIterator(job.getConf(), new IteratorSetting(1, "WholeRow", WholeRowIterator.class));
-    AccumuloInputFormat.addIterator(job.getConf(), new IteratorSetting(2, "Versions", "org.apache.accumulo.core.iterators.VersioningIterator"));
+    Job job = MapreduceWrapper.wrappedJob(bspJob);
+    AccumuloInputFormat.addIterator(job, new IteratorSetting(1, "WholeRow", WholeRowIterator.class));
+    AccumuloInputFormat.addIterator(job, new IteratorSetting(2, "Versions", "org.apache.accumulo.core.iterators.VersioningIterator"));
     IteratorSetting iter = new IteratorSetting(3, "Count", "org.apache.accumulo.core.iterators.CountingIterator");
     iter.addOption("v1", "1");
     iter.addOption("junk", "\0omg:!\\xyzzy");
-    AccumuloInputFormat.addIterator(job.getConf(), iter);
+    AccumuloInputFormat.addIterator(job, iter);
     
-    List<AccumuloIterator> list = AccumuloInputFormat.getIterators(job.getConf());
+    TaskAttemptContext context = MapreduceWrapper.wrappedTaskAttemptContext(bspJob);
+    List<IteratorSetting> list = AccumuloInputFormat.getIterators(context);
     
     // Check the list size
     assertTrue(list.size() == 3);
     
     // Walk the list and make sure our settings are correct
-    AccumuloIterator setting = list.get(0);
+    IteratorSetting setting = list.get(0);
     assertEquals(1, setting.getPriority());
     assertEquals("org.apache.accumulo.core.iterators.user.WholeRowIterator", setting.getIteratorClass());
-    assertEquals("WholeRow", setting.getIteratorName());
+    assertEquals("WholeRow", setting.getName());
     
     setting = list.get(1);
     assertEquals(2, setting.getPriority());
     assertEquals("org.apache.accumulo.core.iterators.VersioningIterator", setting.getIteratorClass());
-    assertEquals("Versions", setting.getIteratorName());
+    assertEquals("Versions", setting.getName());
     
     setting = list.get(2);
     assertEquals(3, setting.getPriority());
     assertEquals("org.apache.accumulo.core.iterators.CountingIterator", setting.getIteratorClass());
-    assertEquals("Count", setting.getIteratorName());
+    assertEquals("Count", setting.getName());
     
-    List<AccumuloIteratorOption> iteratorOptions = AccumuloInputFormat.getIteratorOptions(job.getConf());
+    Map<String,String> iteratorOptions = setting.getOptions();
     assertEquals(2, iteratorOptions.size());
-    assertEquals("Count", iteratorOptions.get(0).getIteratorName());
-    assertEquals("Count", iteratorOptions.get(1).getIteratorName());
-    assertEquals("v1", iteratorOptions.get(0).getKey());
-    assertEquals("1", iteratorOptions.get(0).getValue());
-    assertEquals("junk", iteratorOptions.get(1).getKey());
-    assertEquals("\0omg:!\\xyzzy", iteratorOptions.get(1).getValue());
+    assertTrue(iteratorOptions.containsKey("v1"));
+    assertEquals("1", iteratorOptions.get("v1"));
+    assertTrue(iteratorOptions.containsKey("junk"));
+    assertEquals("\0omg:!\\xyzzy", iteratorOptions.get("junk"));
   }
   
   @Test
-  public void testIteratorOptionEncoding() throws Throwable {
+  public void testIteratorOptionEncoding() throws IOException {
+    BSPJob bspJob = new BSPJob();
     String key = "colon:delimited:key";
     String value = "comma,delimited,value";
+    
+    Job job = MapreduceWrapper.wrappedJob(bspJob);
     IteratorSetting someSetting = new IteratorSetting(1, "iterator", "Iterator.class");
     someSetting.addOption(key, value);
-    BSPJob job = new BSPJob();
-    AccumuloInputFormat.addIterator(job.getConf(), someSetting);
+    AccumuloInputFormat.addIterator(job, someSetting);
     
-    final String rawConfigOpt = new AccumuloIteratorOption("iterator", key, value).toString();
-    
-    assertEquals(rawConfigOpt, job.getConf().get("AccumuloInputFormat.iterators.options"));
-    
-    List<AccumuloIteratorOption> opts = AccumuloInputFormat.getIteratorOptions(job.getConf());
+    TaskAttemptContext context = MapreduceWrapper.wrappedTaskAttemptContext(bspJob);
+    List<IteratorSetting> iters = AccumuloInputFormat.getIterators(context);
+    assertEquals(1, iters.size());
+    assertEquals("iterator", iters.get(0).getName());
+    assertEquals("Iterator.class", iters.get(0).getIteratorClass());
+    assertEquals(1, iters.get(0).getPriority());
+    Map<String,String> opts = iters.get(0).getOptions();
     assertEquals(1, opts.size());
-    assertEquals(opts.get(0).getKey(), key);
-    assertEquals(opts.get(0).getValue(), value);
+    assertTrue(opts.containsKey(key));
+    assertEquals(value, opts.get(key));
     
     someSetting.addOption(key + "2", value);
     someSetting.setPriority(2);
     someSetting.setName("it2");
-    AccumuloInputFormat.addIterator(job.getConf(), someSetting);
-    opts = AccumuloInputFormat.getIteratorOptions(job.getConf());
-    assertEquals(3, opts.size());
-    for (AccumuloIteratorOption opt : opts) {
-      assertEquals(opt.getKey().substring(0, key.length()), key);
-      assertEquals(opt.getValue(), value);
-    }
+    AccumuloInputFormat.addIterator(job, someSetting);
+    
+    context = MapreduceWrapper.wrappedTaskAttemptContext(bspJob);
+    iters = AccumuloInputFormat.getIterators(context);
+    assertEquals(2, iters.size());
+    assertEquals("iterator", iters.get(0).getName());
+    assertEquals("Iterator.class", iters.get(0).getIteratorClass());
+    assertEquals(1, iters.get(0).getPriority());
+    opts = iters.get(0).getOptions();
+    assertEquals(1, opts.size());
+    assertTrue(opts.containsKey(key));
+    assertEquals(value, opts.get(key));
+    assertEquals("it2", iters.get(1).getName());
+    assertEquals("Iterator.class", iters.get(1).getIteratorClass());
+    assertEquals(2, iters.get(1).getPriority());
+    opts = iters.get(1).getOptions();
+    assertEquals(2, opts.size());
+    assertTrue(opts.containsKey(key));
+    assertEquals(value, opts.get(key));
+    assertTrue(opts.containsKey(key + "2"));
+    assertEquals(value, opts.get(key + "2"));
   }
   
   @Test
   public void testGetIteratorSettings() throws IOException {
-    BSPJob job = new BSPJob();
+    BSPJob bspJob = new BSPJob();
+    Job job = MapreduceWrapper.wrappedJob(bspJob);
     
-    AccumuloInputFormat.addIterator(job.getConf(), new IteratorSetting(1, "WholeRow", "org.apache.accumulo.core.iterators.WholeRowIterator"));
-    AccumuloInputFormat.addIterator(job.getConf(), new IteratorSetting(2, "Versions", "org.apache.accumulo.core.iterators.VersioningIterator"));
-    AccumuloInputFormat.addIterator(job.getConf(), new IteratorSetting(3, "Count", "org.apache.accumulo.core.iterators.CountingIterator"));
+    AccumuloInputFormat.addIterator(job, new IteratorSetting(1, "WholeRow", "org.apache.accumulo.core.iterators.WholeRowIterator"));
+    AccumuloInputFormat.addIterator(job, new IteratorSetting(2, "Versions", "org.apache.accumulo.core.iterators.VersioningIterator"));
+    AccumuloInputFormat.addIterator(job, new IteratorSetting(3, "Count", "org.apache.accumulo.core.iterators.CountingIterator"));
     
-    List<AccumuloIterator> list = AccumuloInputFormat.getIterators(job.getConf());
+    TaskAttemptContext context = MapreduceWrapper.wrappedTaskAttemptContext(bspJob);
+    List<IteratorSetting> list = AccumuloInputFormat.getIterators(context);
     
     // Check the list size
-    assertTrue(list.size() == 3);
+    assertEquals(3, list.size());
     
     // Walk the list and make sure our settings are correct
-    AccumuloIterator setting = list.get(0);
+    IteratorSetting setting = list.get(0);
     assertEquals(1, setting.getPriority());
     assertEquals("org.apache.accumulo.core.iterators.WholeRowIterator", setting.getIteratorClass());
-    assertEquals("WholeRow", setting.getIteratorName());
+    assertEquals("WholeRow", setting.getName());
     
     setting = list.get(1);
     assertEquals(2, setting.getPriority());
     assertEquals("org.apache.accumulo.core.iterators.VersioningIterator", setting.getIteratorClass());
-    assertEquals("Versions", setting.getIteratorName());
+    assertEquals("Versions", setting.getName());
     
     setting = list.get(2);
     assertEquals(3, setting.getPriority());
     assertEquals("org.apache.accumulo.core.iterators.CountingIterator", setting.getIteratorClass());
-    assertEquals("Count", setting.getIteratorName());
-    
+    assertEquals("Count", setting.getName());
   }
   
   @Test
   public void testSetRegex() throws IOException {
-    BSPJob job = new BSPJob();
+    BSPJob bspJob = new BSPJob();
+    Job job = MapreduceWrapper.wrappedJob(bspJob);
     
     String regex = ">\"*%<>\'\\";
     
     IteratorSetting is = new IteratorSetting(50, regex, RegExFilter.class);
     RegExFilter.setRegexs(is, regex, null, null, null, false);
-    AccumuloInputFormat.addIterator(job.getConf(), is);
+    AccumuloInputFormat.addIterator(job, is);
     
-    assertTrue(regex.equals(AccumuloInputFormat.getIterators(job.getConf()).get(0).getIteratorName()));
+    TaskAttemptContext context = MapreduceWrapper.wrappedTaskAttemptContext(bspJob);
+    assertEquals(regex, AccumuloInputFormat.getIterators(context).get(0).getName());
   }
   
-  static class TestBSP extends BSP<Key,Value,Key,Value> {
-    Key key = null;
-    int count = 0;
-    
-    @Override
-    public void bsp(BSPPeer<Key,Value,Key,Value> peer) throws IOException, SyncException,
InterruptedException {
-      // this method reads the next key value record from file
-      KeyValuePair<Key,Value> pair;
-      
-      while ((pair = peer.readNext()) != null) {
-        if (key != null) {
-          assertEquals(key.getRow().toString(), new String(pair.getValue().get()));
-        }
-        
-        assertEquals(pair.getKey().getRow(), new Text(String.format("%09x", count + 1)));
-        assertEquals(new String(pair.getValue().get()), String.format("%09x", count));
-        count++;
-        
-        key = new Key(pair.getKey());
-      }
-      
-      peer.sync();
-      assertEquals(100, count);
-    }
-  }
-  
-  @Test
-  public void testBsp() throws Exception {
-    MockInstance mockInstance = new MockInstance("testmapinstance");
-    Connector c = mockInstance.getConnector("root", new byte[] {});
-    if (c.tableOperations().exists("testtable"))
-      c.tableOperations().delete("testtable");
-    c.tableOperations().create("testtable");
-    
-    BatchWriter bw = c.createBatchWriter("testtable", 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();
-    
-    BSPJob job = new BSPJob(new HamaConfiguration());
-    job.setInputFormat(AccumuloInputFormat.class);
-    job.setBspClass(TestBSP.class);
-    job.setInputPath(new Path("test"));
-    AccumuloInputFormat.setInputInfo(job.getConf(), "root", "".getBytes(), "testtable", new
Authorizations());
-    AccumuloInputFormat.setMockInstance(job.getConf(), "testmapinstance");
-    
-    AccumuloInputFormat input = new AccumuloInputFormat();
-    InputSplit[] splits = input.getSplits(job, 0);
-    assertEquals(splits.length, 1);
-    
-    job.waitForCompletion(false);
-  }
 }

Copied: accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatIT.java
(from r1431766, accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java)
URL: http://svn.apache.org/viewvc/accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatIT.java?p2=accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatIT.java&p1=accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java&r1=1431766&r2=1433751&rev=1433751&view=diff
==============================================================================
--- accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatTest.java
(original)
+++ accumulo/contrib/bsp/trunk/src/test/java/org/apache/accumulo/core/client/mapreduce/AccumuloOutputFormatIT.java
Tue Jan 15 23:58:38 2013
@@ -19,6 +19,7 @@ package org.apache.accumulo.core.client.
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 import java.io.IOException;
 import java.util.Iterator;
@@ -26,7 +27,9 @@ import java.util.Map.Entry;
 
 import org.apache.accumulo.bsp.AccumuloInputFormat;
 import org.apache.accumulo.bsp.AccumuloOutputFormat;
+import org.apache.accumulo.bsp.MapreduceWrapper;
 import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
 import org.apache.accumulo.core.client.Connector;
 import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.client.mock.MockInstance;
@@ -37,6 +40,8 @@ import org.apache.accumulo.core.security
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.Job;
 import org.apache.hama.HamaConfiguration;
 import org.apache.hama.bsp.BSP;
 import org.apache.hama.bsp.BSPJob;
@@ -46,14 +51,17 @@ import org.apache.hama.bsp.sync.SyncExce
 import org.apache.hama.util.KeyValuePair;
 import org.junit.Test;
 
-public class AccumuloOutputFormatTest {
+/**
+ * 
+ */
+public class AccumuloOutputFormatIT {
   
-  static class TestBSP extends BSP<Key,Value,Text,Mutation> {
+  static class OutputFormatTestBSP<M extends Writable> extends BSP<Key,Value,Text,Mutation,M>
{
     Key key = null;
     int count = 0;
     
     @Override
-    public void bsp(BSPPeer<Key,Value,Text,Mutation> peer) throws IOException, SyncException,
InterruptedException {
+    public void bsp(BSPPeer<Key,Value,Text,Mutation,M> peer) throws IOException, SyncException,
InterruptedException {
       // this method reads the next key value record from file
       KeyValuePair<Key,Value> pair;
       
@@ -73,7 +81,7 @@ public class AccumuloOutputFormatTest {
     }
     
     @Override
-    public void cleanup(BSPPeer<Key,Value,Text,Mutation> peer) throws IOException {
+    public void cleanup(BSPPeer<Key,Value,Text,Mutation,M> peer) throws IOException
{
       Mutation m = new Mutation("total");
       m.put("", "", Integer.toString(count));
       peer.write(new Text("testtable2"), m);
@@ -81,7 +89,7 @@ public class AccumuloOutputFormatTest {
   }
   
   @Test
-  public void testBSP() throws Exception {
+  public void testBSPOutputFormat() throws Exception {
     MockInstance mockInstance = new MockInstance("testmrinstance");
     Connector c = mockInstance.getConnector("root", new byte[] {});
     if (c.tableOperations().exists("testtable1"))
@@ -91,7 +99,7 @@ public class AccumuloOutputFormatTest {
     
     c.tableOperations().create("testtable1");
     c.tableOperations().create("testtable2");
-    BatchWriter bw = c.createBatchWriter("testtable1", 10000L, 1000L, 4);
+    BatchWriter bw = c.createBatchWriter("testtable1", new BatchWriterConfig());
     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()));
@@ -100,29 +108,33 @@ public class AccumuloOutputFormatTest {
     bw.close();
     
     Configuration conf = new Configuration();
-    BSPJob bsp = new BSPJob(new HamaConfiguration(conf));
-    bsp.setJobName("Test Input Output");
+    BSPJob bspJob = new BSPJob(new HamaConfiguration(conf));
+    bspJob.setJobName("Test Input Output");
     
-    bsp.setBspClass(TestBSP.class);
-    bsp.setInputFormat(AccumuloInputFormat.class);
-    bsp.setInputPath(new Path("test"));
-    
-    bsp.setOutputFormat(AccumuloOutputFormat.class);
-    bsp.setOutputPath(new Path("test"));
-    
-    bsp.setOutputKeyClass(Text.class);
-    bsp.setOutputValueClass(Mutation.class);
-    
-    AccumuloInputFormat.setInputInfo(bsp.getConf(), "root", "".getBytes(), "testtable1",
new Authorizations());
-    AccumuloInputFormat.setMockInstance(bsp.getConf(), "testmrinstance");
-    AccumuloOutputFormat.setOutputInfo(bsp.getConf(), "root", "".getBytes(), false, "testtable2");
-    AccumuloOutputFormat.setMockInstance(bsp.getConf(), "testmrinstance");
+    bspJob.setBspClass(OutputFormatTestBSP.class);
+    bspJob.setInputFormat(AccumuloInputFormat.class);
+    bspJob.setInputPath(new Path("test"));
+    
+    bspJob.setOutputFormat(AccumuloOutputFormat.class);
+    bspJob.setJar("target/integration-tests.jar");
+    bspJob.setOutputPath(new Path("target/bsp-outputformat-test"));
+    
+    bspJob.setOutputKeyClass(Text.class);
+    bspJob.setOutputValueClass(Mutation.class);
+    
+    Job job = MapreduceWrapper.wrappedJob(bspJob);
+    
+    AccumuloInputFormat.setInputInfo(job, "root", "".getBytes(), "testtable1", new Authorizations());
+    AccumuloInputFormat.setMockInstance(job, "testmrinstance");
+    AccumuloOutputFormat.setOutputInfo(job, "root", "".getBytes(), false, "testtable2");
+    AccumuloOutputFormat.setMockInstance(job, "testmrinstance");
     
     AccumuloInputFormat input = new AccumuloInputFormat();
-    InputSplit[] splits = input.getSplits(bsp, 0);
+    InputSplit[] splits = input.getSplits(bspJob, 0);
     assertEquals(splits.length, 1);
     
-    bsp.waitForCompletion(false);
+    if (!bspJob.waitForCompletion(false))
+      fail("Job not finished successfully");
     
     Scanner scanner = c.createScanner("testtable2", new Authorizations());
     Iterator<Entry<Key,Value>> iter = scanner.iterator();
@@ -131,6 +143,6 @@ public class AccumuloOutputFormatTest {
     assertEquals("total", entry.getKey().getRow().toString());
     assertEquals(100, Integer.parseInt(new String(entry.getValue().get())));
     assertFalse(iter.hasNext());
-    
   }
+  
 }



Mime
View raw message