Repository: apache/incubator-fluo
Pull request #704 (2016-07-13 07:36:18)

Fixes435
@keith-turner @ctubbsii 
Removed all unwanted dependencies from pom which are not used.

keith-turner: I think commons config can be removed from all poms except for the api module.  I experimented with making that chane in this branch and something really strange happened.  When I ran dependency analyze, it said the following for core.

```
[WARNING] Used undeclared dependencies found:
[WARNING]    log4j:log4j:jar:1.2.16:compile
[WARNING]    commons-configuration:commons-configuration:jar:1.10:compile
[WARNING]    org.apache.hadoop:hadoop-common:jar:2.6.3:compile
[WARNING] Unused declared dependencies found:
[WARNING]    org.apache.hadoop:hadoop-client:jar:2.6.3:compile
```

I dont understand why this is happening because core does not directly use commons config.  When I run the following it shows that only API directly uses commons config.

```
$ find . -name "*.java" | xargs grep commons.conf
./modules/api/src/main/java/org/apache/fluo/api/config/SimpleConfiguration.java:import org.apache.commons.configuration.CompositeConfiguration;
./modules/api/src/main/java/org/apache/fluo/api/config/SimpleConfiguration.java:import org.apache.commons.configuration.Configuration;
./modules/api/src/main/java/org/apache/fluo/api/config/SimpleConfiguration.java:import org.apache.commons.configuration.ConfigurationException;
./modules/api/src/main/java/org/apache/fluo/api/config/SimpleConfiguration.java:import org.apache.commons.configuration.ConfigurationUtils;
./modules/api/src/main/java/org/apache/fluo/api/config/SimpleConfiguration.java:import org.apache.commons.configuration.PropertiesConfiguration;
```

Does anyone else know what is going on?
keith-turner: @Garvit244 I see some additional changes that can be made.

 * For api module can remove commons collections.
 * For integration module can add zookeeper, log4j, and guava at test scope
 * For cluster module can add javax inject
 * For cluster module can remove fluo-mini and fluo-integration

Still need to figure out whats going on with commons configuration.

For the core module something strange is also going on with log4j.  Its used in the test scope, but dependency:analyze says it need to be added to the compile scope.  I wonder if we added it to the test scope if it would make this warning go away.
keith-turner: Looking at the dependency:analyze documentation I noticed it has a `failOnWarning` options.  I think we should add dependency analyze plugin with this option (configuring it to ignore hadoop stuff) to the Fluo poms.  That will help us keep our dependency declarations correct in the future which will benefit users.  @Garvit244 would you be interested in add this as part of this pull request?  If not I can open an issue and we can add it another pull request.
Garvit244: Sure. I will add that in this pull request and will do all required changes
keith-turner: I was looking at the documentation again and noticed the dependency:analyze plugin has an `outputXML` option.  I gave it a try, its really neat.

Garvit244: @keith-turner can you tell me how you check results of these two configuration `outputXML`.
Added this snippet in parent pom code-base :
`<failOnWarning>true</failOnWarning>
              <ignoredDependencies>
                <ignoredDependency>org.apache.hadoop:hadoop-common:jar:2.6.3</ignoredDependency>
                <ignoredDependency>org.apache.hadoop:hadoop-client:jar:2.6.3</ignoredDependency>
                <ignoredDependency>org.apache.hadoop:hadoop-mapreduce-client-core:jar:2.6.3</ignoredDependency>
              </ignoredDependencies>
              <outputXML>true</outputXML>`

Running `mvn verify` to check result and its output but getting error.
Garvit244: Removing dependency from cluster module `fluo-mini`throws exception `project fluo-cluster: Unable to start MiniAccumuloCluster: Initialize process returned 1`
keith-turner: > @keith-turner can you tell me how you check results of these two configuration outputXML and failOnWarning.

For `outputXML` I was just thought it was nice to use at the command line when trying to fix the pom.  But I was not thinking it should be in the pom.  Can use it like

```
mvn dependency:analyze -DoutputXML=true
```

For `failOnWarning` , does it cause the build to fail if you comment out one of the `ignoredDependency`? My hope is that it will fail the build if we add a new dependency in code without adding it to the pom.
keith-turner: > Removing dependency from cluster module fluo-minithrows exception project fluo-cluster: Unable to start MiniAccumuloCluster: Initialize process returned 1

Oh, I think I know whats going on there.   That module used to have integration test, but it no longer does. So it used to need MiniAccumulo, but it no longer does.   In addition to removing the `fluo-mini` and  `fluo-integration` can you also try removing the profile with id `<id>mini-accumulo</id>`?  This profile activates the maven-accumulo plugin that runs MiniAccumulo which is not being used.
Garvit244: It will fail if any of dependency is used or unused without specifying in pom. One need to add that dependency to `<ignoredependency>` tag. 
If in near future any dependency is added then we can keep track whether that dependency is added in pom or not. This will keep our pom clear and keep track of dependencies.  
Garvit244: Q1: `fluo-mapreduce` module uses accumulo-minicluster as dependency for test. Is this dependency needed in fluo-mapreduce pom?
Q2: `fluo-distribution` module `src/main/assembly/bin.xml ` assembly descriptor is readed which was failing with error:
`Execution bin-assembly of goal org.apache.maven.plugins:maven-assembly-plugin:2.6:single failed: group id '1493847943' is too big ( > 2097151 ). Use STAR or POSIX extensions to overcome this limit`
keith-turner: I figured out hoe fluo-core depends on commons config.  Fluo core uses Accumulo's ClientConfiguration class which extends a commons config class.
keith-turner: > Q1: fluo-mapreduce module uses accumulo-minicluster as dependency for test. Is this dependency needed in fluo-mapreduce pom?

The accumulo maven plugin uses it.   I am not sure what the best way to handle this it.  I tried added the following to `mapreduce/pom.xml`.  This made the plugin ignore mini accumulo just for that module.   The `combine.children="append"` setting was key, I just learned about it from looking at some maven docs.  I am not completely sure I used `combine.children` correctly, but it seems to work.

```xml
  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-dependency-plugin</artifactId>
        <executions>
          <execution>
            <id>analyze</id>
            <configuration>
              <ignoredDependencies combine.children="append">
                <ignoredDependency>org.apache.accumulo:accumulo-minicluster:jar:1.7.1</ignoredDependency>
              </ignoredDependencies>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
```
keith-turner: > Q2: fluo-distribution module src/main/assembly/bin.xml assembly descriptor is readed which was failing with error:

I have no idea whats going on with that error.  I am not seeing anything like that.  Maybe your linux user has a really large group id that it can not handle??
keith-turner: @Garvit244 this looks good.  After the `accumulo-minicluster` ignore is added to the mapreduce pom and the commons config stuff is fixed up, I can merge it.
Garvit244: Done
keith-turner: @Garvit244 after those two issues are fixed and the Travis CI build passes, I can merge this.  When I merge it are you ok with me squishing it to one commit? You will still be listed as the author. 
Garvit244: @keith-turner sure will do all changes. My Branch has diverged from master also do i need to merge with master first before doing PR.
keith-turner: > My Branch has diverged from master also do i need to merge with master first before doing PR.

No, we usually do not merge or rebase until the code review is finished.   Rebasing and merging make it harder for the reviewer to follow changes.   After the code review is done and TravisCI builds successfully, then squash to one commit and push that single commit to master.
keith-turner: +1  

Thanks @Garvit244 this looks good and the build passed.  You have three commits two of them are merge commits, can you squash this branch into one commit?





Patch:
diff --git a/docs/applications.md b/docs/applications.md
index ceae1d60..8e76a024 100644
--- a/docs/applications.md
+++ b/docs/applications.md
@@ -4,7 +4,7 @@ Creating Fluo applications
 Once you have Fluo installed and running on your cluster, you can now run Fluo applications which consist of 
 clients and observers.
 
-If you are new to Fluo, consider first running the [phrasecount] application on your Fluo instance. Otherwise,
+If you are new to Fluo, consider first running the [phrasecount] application on your Fluo instance.  Otherwise,
 you can create your own Fluo client or observer by the following the steps below.
  
 For both clients and observers, you will need to include the following in your Maven pom:
@@ -93,7 +93,7 @@ To create an observer, follow these steps:
 
 1. Create a class that extends [AbstractObserver].
 2. Build a jar containing this class and include this jar in the `lib/` directory of your Fluo application.
-3. Configure your Fluo instance to use this observer by modifying the Observer section of [fluo.properties].
+3. Configure your Fluo instance to use this observer by modifying the Observer section of [fluo.properties].  
 4. Restart your Fluo instance so that your Fluo workers load the new observer.
 
 Application Configuration
@@ -114,6 +114,7 @@ provide enough information to debug the root cause of the problem.  To help
 debug Fluo applications, low-level logging of transactions can be turned on by
 setting the following loggers to TRACE:
 
+
 | Logger               | Level | Information                                                                                        |
 |----------------------|-------|----------------------------------------------------------------------------------------------------|
 | `fluo.tx`            | TRACE | Provides detailed information about what transactions read and wrote                               |
diff --git a/docs/contributing.md b/docs/contributing.md
index 17af2e62..7fb7c0f8 100644
--- a/docs/contributing.md
+++ b/docs/contributing.md
@@ -4,7 +4,7 @@ Contributing to Fluo
 Building Fluo
 -------------
 
-If you have [Git], [Maven], and [Java][java] (version 8+) installed, run these commands
+If you have [Git], [Maven], and [Java](version 7+) installed, run these commands
 to build Fluo:
 
     git clone https://github.com/apache/incubator-fluo.git
@@ -22,6 +22,6 @@ a local Fluo instance (called MiniFluo) and run against it.
 * A [Stress test][Stress] application designed to run on a cluster.
 
 [Git]: http://git-scm.com/
-[java]: http://openjdk.java.net/
+[Java]: https://www.oracle.com/java/index.html
 [Maven]: http://maven.apache.org/
 [Stress]: https://github.com/fluo-io/fluo-stress
diff --git a/docs/grafana.md b/docs/grafana.md
index 5d10b969..3263e8bc 100644
--- a/docs/grafana.md
+++ b/docs/grafana.md
@@ -2,7 +2,7 @@
 # Fluo metrics in Grafana/InfluxDB
 
 Fluo is instrumented using [dropwizard metrics][1] which allows Fluo to be configured
-to send metrics to multiple metrics tools (such as Graphite, Ganglia, etc).
+to send metrics to multiple metrics tools (such as Graphite, Ganglia, etc).  
 
 This document describes how to send Fluo metrics to [InfluxDB], a time series database, and make 
 them viewable in [Grafana], a visualization tool.  If you want general information on metrics, see the 
diff --git a/docs/metrics.md b/docs/metrics.md
index eb2a1e43..b98e68f1 100644
--- a/docs/metrics.md
+++ b/docs/metrics.md
@@ -11,8 +11,8 @@ different tools, below are some tools that have been used with Fluo.
    these tools for you and configure Fluo to send to them.
 
 2. JMX - Fluo can be configured to reports metrics via JMX which can be viewed
-   in jconsole or jvisualvm.
- 
+   in jconsole or jvisualvm.  
+   
 3. CSV - Fluo can be configured to output metrics as CSV to a specified directory.
 
 Configuring Reporters
@@ -20,13 +20,13 @@ Configuring Reporters
 
 Inorder to configure metrics reporters, look at the metrics section in an
 applications `fluo.properties` file.  This sections has a lot of commented out
-options for configuring reporters.
+options for configuring reporters.  
 
     fluo.metrics.reporter.console.enable=false
     fluo.metrics.reporter.console.frequency=30
 
 The frequency is in seconds for all reporters.
-
+  
 Metrics reported by Fluo
 ------------------------
 
@@ -39,7 +39,7 @@ this id consist of a hostname and a base36 long that is unique across all fluo p
 Some of the metrics reported have the class name as the suffix.  This classname
 is the observer or load task that executed the transactions.   This should
 allow things like transaction collisions to be tracked per class.  In the
-table below this is denoted with `<cn>`.
+table below this is denoted with `<cn>`.  
 
 |Metric                                 | Type           | Description                         |
 |---------------------------------------|----------------|-------------------------------------|
diff --git a/docs/prod-fluo-setup.md b/docs/prod-fluo-setup.md
index e21350b9..31027fe8 100644
--- a/docs/prod-fluo-setup.md
+++ b/docs/prod-fluo-setup.md
@@ -215,8 +215,8 @@ distribution on every node in your cluster.
 [Accumulo]: https://accumulo.apache.org/
 [Hadoop]: http://hadoop.apache.org/
 [Zookeeper]: http://zookeeper.apache.org/
-[Java]: http://openjdk.java.net/
-[release]: https://fluo.apache.org/download/
+[Java]: https://www.oracle.com/java/index.html
+[release]: https://github.com/apache/incubator-fluo/releases
 [phrasecount]: https://github.com/fluo-io/phrasecount
 [fluo.properties]: ../modules/distribution/src/main/config/fluo.properties
 [fluo-env.sh]: ../modules/distribution/src/main/config/fluo-env.sh
diff --git a/modules/accumulo/pom.xml b/modules/accumulo/pom.xml
index 55544d2a..9351bc9f 100644
--- a/modules/accumulo/pom.xml
+++ b/modules/accumulo/pom.xml
@@ -62,10 +62,5 @@
       <artifactId>easymock</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-log4j12</artifactId>
-      <scope>test</scope>
-    </dependency>
   </dependencies>
 </project>
diff --git a/modules/api/pom.xml b/modules/api/pom.xml
index 61c8e2a1..83078d7f 100644
--- a/modules/api/pom.xml
+++ b/modules/api/pom.xml
@@ -32,10 +32,6 @@
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>
     </dependency>
-    <dependency>
-      <groupId>commons-collections</groupId>
-      <artifactId>commons-collections</artifactId>
-    </dependency>
     <dependency>
       <groupId>commons-configuration</groupId>
       <artifactId>commons-configuration</artifactId>
@@ -44,21 +40,11 @@
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
     </dependency>
-    <dependency>
-      <groupId>commons-io</groupId>
-      <artifactId>commons-io</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-log4j12</artifactId>
-      <scope>test</scope>
-    </dependency>
   </dependencies>
   <build>
     <plugins>
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java b/modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java
index 79e1cef3..778aa940 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/client/SnapshotBase.java
@@ -19,11 +19,11 @@ import java.util.Collection;
 import java.util.Map;
 import java.util.Set;
 
-import org.apache.fluo.api.client.scanner.ScannerBuilder;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.api.data.RowColumn;
-import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.iterator.RowIterator;
 
 /**
  * Allows users to read from a Fluo table at a certain point in time
@@ -57,44 +57,9 @@ public interface SnapshotBase {
   Map<Bytes, Map<Column, Bytes>> get(Collection<RowColumn> rowColumns);
 
   /**
-   * This method is the starting point for constructing a scanner. Scanners can be constructed over
-   * a {@link Span} and/or with a subset of columns. Below is simple example of building a scanner.
-   * 
-   * <pre>
-   * {@code
-   *   Transaction tx = ...;
-   *   Span span = Span.exact("row4");
-   *   Column col1 = new Column("fam1","qual1");
-   *   Column col2 = new Column("fam1","qual2");
-   * 
-   *   //create a scanner over row4 fetching columns fam1:qual1 and fam1:qual2
-   *   CellScanner cs = tx.scanner().over(span).fetch(col1,col2).build();
-   *   for(RowColumnValue rcv : cs) {
-   *     //do stuff with rcv
-   *   }
-   * }
-   * </pre>
-   *
-   * <p>
-   * The following example shows how to build a row scanner.
-   *
-   * <pre>
-   * {
-   *   &#064;code
-   *   RowScanner rs = tx.scanner().over(span).fetch(col1, col2).byRow().build();
-   *   for (ColumnScanner colScanner : rs) {
-   *     Bytes row = colScanner.getRow();
-   *     for (ColumnValue cv : colScanner) {
-   *       // do stuff with the columns and values in the row
-   *     }
-   *   }
-   * }
-   * </pre>
-   *
-   * @return A scanner builder.
+   * Retrieves a {@link RowIterator} with the given {@link ScannerConfiguration}
    */
-
-  ScannerBuilder scanner();
+  RowIterator get(ScannerConfiguration config);
 
   /**
    * Wrapper for {@link #get(Collection)} that uses Strings. All strings are encoded and decoded
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/ColumnScanner.java b/modules/api/src/main/java/org/apache/fluo/api/client/scanner/ColumnScanner.java
deleted file mode 100644
index dd28ce9f..00000000
--- a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/ColumnScanner.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.fluo.api.client.scanner;
-
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.ColumnValue;
-
-/**
- * @since 1.0.0
- */
-public interface ColumnScanner extends Iterable<ColumnValue> {
-
-  /**
-   * @return the row for all column values
-   */
-  Bytes getRow();
-
-  /**
-   * @return the row for all column values decoded as UTF-8
-   */
-  String getsRow();
-}
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/RowScannerBuilder.java b/modules/api/src/main/java/org/apache/fluo/api/client/scanner/RowScannerBuilder.java
deleted file mode 100644
index fa7c60eb..00000000
--- a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/RowScannerBuilder.java
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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.fluo.api.client.scanner;
-
-/**
- * @since 1.0.0
- */
-public interface RowScannerBuilder {
-  /**
-   * @return a new scanner created with any previously set restrictions
-   */
-  RowScanner build();
-}
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/ScannerBuilder.java b/modules/api/src/main/java/org/apache/fluo/api/client/scanner/ScannerBuilder.java
deleted file mode 100644
index 1217aa0c..00000000
--- a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/ScannerBuilder.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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.fluo.api.client.scanner;
-
-import java.util.Collection;
-
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.Span;
-
-/**
- * @since 1.0.0
- */
-public interface ScannerBuilder {
-  /**
-   * @param span restrict the scanner to data within span
-   * @return self
-   */
-  ScannerBuilder over(Span span);
-
-
-  /**
-   * Passing in a Column with only the family set will fetch the entire column family.
-   *
-   * @param columns restrict the scanner to only these columns
-   * @return self
-   */
-  ScannerBuilder fetch(Column... columns);
-
-  /**
-   * Passing in a Column with only the family set will fetch the entire column family.
-   *
-   * @param columns restrict the scanner to only these columns
-   * @return self
-   */
-  ScannerBuilder fetch(Collection<Column> columns);
-
-  /**
-   * @return a new scanner created with any previously set restrictions
-   */
-  CellScanner build();
-
-  /**
-   * Call this to build a row scanner.
-   *
-   * @return a row scanner builder using any previously set restrictions
-   */
-  RowScannerBuilder byRow();
-}
diff --git a/modules/api/src/main/java/org/apache/fluo/api/config/ScannerConfiguration.java b/modules/api/src/main/java/org/apache/fluo/api/config/ScannerConfiguration.java
new file mode 100644
index 00000000..3a73b508
--- /dev/null
+++ b/modules/api/src/main/java/org/apache/fluo/api/config/ScannerConfiguration.java
@@ -0,0 +1,98 @@
+/*
+ * 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.fluo.api.config;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Objects;
+import java.util.Set;
+
+import org.apache.fluo.api.client.SnapshotBase;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.Span;
+
+/**
+ * Contains configuration for a {@link org.apache.fluo.api.client.Snapshot} scanner. Passed to
+ * {@link SnapshotBase#get(ScannerConfiguration)}.
+ *
+ * @since 1.0.0
+ */
+public class ScannerConfiguration implements Cloneable {
+
+  private Span span = new Span();
+  private Set<Column> columns = new HashSet<>();
+
+  /**
+   * Sets {@link Span} for ScannerConfiguration
+   */
+  public ScannerConfiguration setSpan(Span span) {
+    Objects.requireNonNull(span);
+    this.span = span;
+    return this;
+  }
+
+  /**
+   * Retrieves {@link Span} for ScannerConfiguration
+   */
+  public Span getSpan() {
+    return span;
+  }
+
+  /**
+   * List of all {@link Column}s that scanner will retrieve
+   */
+  public Set<Column> getColumns() {
+    return Collections.unmodifiableSet(columns);
+  }
+
+  /**
+   * Configures scanner to retrieve column with the given family
+   */
+  public ScannerConfiguration fetchColumnFamily(Bytes fam) {
+    Objects.requireNonNull(fam);
+    columns.add(new Column(fam));
+    return this;
+  }
+
+  /**
+   * Configures scanner to retrieve column with the given family and qualifier
+   */
+  public ScannerConfiguration fetchColumn(Bytes fam, Bytes qual) {
+    Objects.requireNonNull(fam);
+    Objects.requireNonNull(qual);
+    columns.add(new Column(fam, qual));
+    return this;
+  }
+
+  /**
+   * Clears all fetched column settings
+   */
+  public void clearColumns() {
+    columns.clear();
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public Object clone() throws CloneNotSupportedException {
+    ScannerConfiguration sc = (ScannerConfiguration) super.clone();
+
+    sc.columns = (Set<Column>) ((HashSet<Column>) columns).clone();
+    sc.span = span;
+
+    return sc;
+  }
+}
diff --git a/modules/api/src/main/java/org/apache/fluo/api/data/ColumnValue.java b/modules/api/src/main/java/org/apache/fluo/api/data/ColumnValue.java
deleted file mode 100644
index 4080e8c6..00000000
--- a/modules/api/src/main/java/org/apache/fluo/api/data/ColumnValue.java
+++ /dev/null
@@ -1,83 +0,0 @@
-/*
- * 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.fluo.api.data;
-
-import java.io.Serializable;
-
-/**
- * @since 1.0.0
- */
-
-public class ColumnValue implements Serializable, Comparable<ColumnValue> {
-  private static final long serialVersionUID = 1L;
-
-  private Column column;
-  private Bytes val;
-
-  public ColumnValue(Column col, Bytes val) {
-    this.column = col;
-    this.val = val;
-  }
-
-  public ColumnValue(Column col, String val) {
-    this.column = col;
-    this.val = Bytes.of(val);
-  }
-
-  public Column getColumn() {
-    return column;
-  }
-
-  public Bytes getValue() {
-    return val;
-  }
-
-  /**
-   * @return value as UTF-8 decoded string
-   */
-  public String getsValue() {
-    return val.toString();
-  }
-
-  @Override
-  public int compareTo(ColumnValue o) {
-    int comp = column.compareTo(o.column);
-    if (comp == 0) {
-      comp = val.compareTo(o.val);
-    }
-    return comp;
-  }
-
-  @Override
-  public boolean equals(Object o) {
-    if (o instanceof ColumnValue) {
-      ColumnValue ocv = (ColumnValue) o;
-      return column.equals(ocv.column) && val.equals(ocv.val);
-    }
-
-    return false;
-  }
-
-  @Override
-  public int hashCode() {
-    return column.hashCode() + 31 * val.hashCode();
-  }
-
-  @Override
-  public String toString() {
-    return column + " " + val;
-  }
-}
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/RowScanner.java b/modules/api/src/main/java/org/apache/fluo/api/iterator/ColumnIterator.java
similarity index 74%
rename from modules/api/src/main/java/org/apache/fluo/api/client/scanner/RowScanner.java
rename to modules/api/src/main/java/org/apache/fluo/api/iterator/ColumnIterator.java
index e65816b1..be04738e 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/RowScanner.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/iterator/ColumnIterator.java
@@ -13,11 +13,19 @@
  * the License.
  */
 
-package org.apache.fluo.api.client.scanner;
+package org.apache.fluo.api.iterator;
+
+import java.util.Iterator;
+import java.util.Map.Entry;
+
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
 
 /**
+ * Iterator for Fluo columns
+ *
  * @since 1.0.0
  */
-public interface RowScanner extends Iterable<ColumnScanner> {
+public interface ColumnIterator extends Iterator<Entry<Column, Bytes>> {
 
 }
diff --git a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/CellScanner.java b/modules/api/src/main/java/org/apache/fluo/api/iterator/RowIterator.java
similarity index 77%
rename from modules/api/src/main/java/org/apache/fluo/api/client/scanner/CellScanner.java
rename to modules/api/src/main/java/org/apache/fluo/api/iterator/RowIterator.java
index b2bf50ec..21e53b13 100644
--- a/modules/api/src/main/java/org/apache/fluo/api/client/scanner/CellScanner.java
+++ b/modules/api/src/main/java/org/apache/fluo/api/iterator/RowIterator.java
@@ -13,13 +13,18 @@
  * the License.
  */
 
-package org.apache.fluo.api.client.scanner;
+package org.apache.fluo.api.iterator;
 
-import org.apache.fluo.api.data.RowColumnValue;
+import java.util.Iterator;
+import java.util.Map.Entry;
+
+import org.apache.fluo.api.data.Bytes;
 
 /**
+ * Iterator for Fluo rows
+ *
  * @since 1.0.0
  */
-public interface CellScanner extends Iterable<RowColumnValue> {
+public interface RowIterator extends Iterator<Entry<Bytes, ColumnIterator>> {
 
 }
diff --git a/modules/cluster/pom.xml b/modules/cluster/pom.xml
index 07e433fc..cf2a75b7 100644
--- a/modules/cluster/pom.xml
+++ b/modules/cluster/pom.xml
@@ -51,12 +51,9 @@
       <version>4.0</version>
     </dependency>
     <dependency>
-      <groupId>commons-configuration</groupId>
-      <artifactId>commons-configuration</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>commons-io</groupId>
-      <artifactId>commons-io</artifactId>
+      <groupId>javax.inject</groupId>
+      <artifactId>javax.inject</artifactId>
+      <version>1</version>
     </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
@@ -66,10 +63,6 @@
       <groupId>org.apache.curator</groupId>
       <artifactId>curator-framework</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.curator</groupId>
-      <artifactId>curator-recipes</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.fluo</groupId>
       <artifactId>fluo-accumulo</artifactId>
@@ -82,10 +75,6 @@
       <groupId>org.apache.fluo</groupId>
       <artifactId>fluo-core</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.apache.fluo</groupId>
-      <artifactId>fluo-mini</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client</artifactId>
@@ -118,29 +107,5 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.apache.fluo</groupId>
-      <artifactId>fluo-integration</artifactId>
-      <type>test-jar</type>
-      <scope>test</scope>
-    </dependency>
   </dependencies>
-  <profiles>
-    <profile>
-      <id>mini-accumulo</id>
-      <activation>
-        <property>
-          <name>!skipTests</name>
-        </property>
-      </activation>
-      <build>
-        <plugins>
-          <plugin>
-            <groupId>org.apache.accumulo</groupId>
-            <artifactId>accumulo-maven-plugin</artifactId>
-          </plugin>
-        </plugins>
-      </build>
-    </profile>
-  </profiles>
 </project>
diff --git a/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/AppRunner.java b/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/AppRunner.java
index d40ff9ec..ca9afe0d 100644
--- a/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/AppRunner.java
+++ b/modules/cluster/src/main/java/org/apache/fluo/cluster/runner/AppRunner.java
@@ -17,8 +17,7 @@ package org.apache.fluo.cluster.runner;
 
 import java.lang.reflect.Method;
 import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashSet;
+import java.util.Map;
 
 import javax.inject.Provider;
 
@@ -36,13 +35,14 @@ import org.apache.fluo.accumulo.format.FluoFormatter;
 import org.apache.fluo.api.client.FluoClient;
 import org.apache.fluo.api.client.FluoFactory;
 import org.apache.fluo.api.client.Snapshot;
-import org.apache.fluo.api.client.scanner.CellScanner;
 import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumnValue;
 import org.apache.fluo.api.data.Span;
 import org.apache.fluo.api.exceptions.FluoException;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.cluster.util.FluoYarnConfig;
 import org.apache.fluo.core.impl.Environment;
 import org.apache.fluo.core.impl.Notification;
@@ -68,8 +68,9 @@ public abstract class AppRunner {
     this.scriptName = scriptName;
   }
 
-  public static Span getSpan(ScanOptions options) {
-    Span span = new Span();
+  public static ScannerConfiguration buildScanConfig(ScanOptions options) {
+    ScannerConfiguration scanConfig = new ScannerConfiguration();
+
     if ((options.getExactRow() != null)
         && ((options.getStartRow() != null) || (options.getEndRow() != null) || (options
             .getRowPrefix() != null))) {
@@ -86,43 +87,35 @@ public abstract class AppRunner {
 
     // configure span of scanner
     if (options.getExactRow() != null) {
-      span = Span.exact(options.getExactRow());
+      scanConfig.setSpan(Span.exact(options.getExactRow()));
     } else if (options.getRowPrefix() != null) {
-      span = Span.prefix(options.getRowPrefix());
+      scanConfig.setSpan(Span.prefix(options.getRowPrefix()));
     } else {
       if ((options.getStartRow() != null) && (options.getEndRow() != null)) {
-        span = new Span(options.getStartRow(), true, options.getEndRow(), true);
+        scanConfig.setSpan(new Span(options.getStartRow(), true, options.getEndRow(), true));
       } else if (options.getStartRow() != null) {
-        span = new Span(Bytes.of(options.getStartRow()), true, Bytes.EMPTY, true);
+        scanConfig.setSpan(new Span(Bytes.of(options.getStartRow()), true, Bytes.EMPTY, true));
       } else if (options.getEndRow() != null) {
-        span = new Span(Bytes.EMPTY, true, Bytes.of(options.getEndRow()), true);
+        scanConfig.setSpan(new Span(Bytes.EMPTY, true, Bytes.of(options.getEndRow()), true));
       }
     }
 
-    return span;
-  }
-
-  public static Collection<Column> getColumns(ScanOptions options) {
-    Collection<Column> columns = new HashSet<>();
-
     // configure columns of scanner
     for (String column : options.getColumns()) {
       String[] colFields = column.split(":");
       if (colFields.length == 1) {
-        columns.add(new Column(colFields[0]));
+        scanConfig.fetchColumnFamily(Bytes.of(colFields[0]));
       } else if (colFields.length == 2) {
-        columns.add(new Column(colFields[0], colFields[1]));
+        scanConfig.fetchColumn(Bytes.of(colFields[0]), Bytes.of(colFields[1]));
       } else {
         throw new IllegalArgumentException("Failed to scan!  Column '" + column
             + "' has too many fields (indicated by ':')");
       }
     }
 
-    return columns;
+    return scanConfig;
   }
 
-
-
   public long scan(FluoConfiguration config, String[] args) {
     ScanOptions options = new ScanOptions();
     JCommander jcommand = new JCommander(options);
@@ -155,45 +148,45 @@ public abstract class AppRunner {
     try (FluoClient client = FluoFactory.newClient(sConfig)) {
       try (Snapshot s = client.newSnapshot()) {
 
-        Span span = null;
-        Collection<Column> columns = null;
+        ScannerConfiguration scanConfig = null;
         try {
-          span = getSpan(options);
-          columns = getColumns(options);
+          scanConfig = buildScanConfig(options);
         } catch (IllegalArgumentException e) {
           System.err.println(e.getMessage());
           System.exit(-1);
         }
 
-        CellScanner cellScanner = s.scanner().over(span).fetch(columns).build();
+        RowIterator iter = s.get(scanConfig);
 
-        StringBuilder sb = new StringBuilder();
-        for (RowColumnValue rcv : cellScanner) {
-          if (options.hexEncNonAscii) {
-            sb.setLength(0);
-            Hex.encNonAscii(sb, rcv.getRow());
-            sb.append(" ");
-            Hex.encNonAscii(sb, rcv.getColumn(), " ");
-            sb.append("\t");
-            Hex.encNonAscii(sb, rcv.getValue());
-            System.out.println(sb.toString());
-          } else {
-            sb.setLength(0);
-            sb.append(rcv.getsRow());
-            sb.append(" ");
-            sb.append(rcv.getColumn());
-            sb.append("\t");
-            sb.append(rcv.getsValue());
-            System.out.println(sb.toString());
-          }
-          entriesFound++;
-          if (System.out.checkError()) {
-            break;
-          }
+        if (!iter.hasNext()) {
+          System.out.println("\nNo data found\n");
         }
 
-        if (entriesFound == 0) {
-          System.out.println("\nNo data found\n");
+        StringBuilder sb = new StringBuilder();
+        while (iter.hasNext() && !System.out.checkError()) {
+          Map.Entry<Bytes, ColumnIterator> rowEntry = iter.next();
+          ColumnIterator citer = rowEntry.getValue();
+          while (citer.hasNext() && !System.out.checkError()) {
+            Map.Entry<Column, Bytes> colEntry = citer.next();
+            if (options.hexEncNonAscii) {
+              sb.setLength(0);
+              Hex.encNonAscii(sb, rowEntry.getKey());
+              sb.append(" ");
+              Hex.encNonAscii(sb, colEntry.getKey(), " ");
+              sb.append("\t");
+              Hex.encNonAscii(sb, colEntry.getValue());
+              System.out.println(sb.toString());
+            } else {
+              sb.setLength(0);
+              sb.append(rowEntry.getKey());
+              sb.append(" ");
+              sb.append(colEntry.getKey());
+              sb.append("\t");
+              sb.append(colEntry.getValue());
+              System.out.println(sb.toString());
+            }
+            entriesFound++;
+          }
         }
       } catch (FluoException e) {
         System.out.println("Scan failed - " + e.getMessage());
@@ -208,11 +201,9 @@ public abstract class AppRunner {
 
     Connector conn = AccumuloUtil.getConnector(sConfig);
 
-    Span span = null;
-    Collection<Column> columns = null;
+    ScannerConfiguration scanConfig = null;
     try {
-      span = getSpan(options);
-      columns = getColumns(options);
+      scanConfig = buildScanConfig(options);
     } catch (IllegalArgumentException e) {
       System.err.println(e.getMessage());
       System.exit(-1);
@@ -222,8 +213,8 @@ public abstract class AppRunner {
 
     try {
       Scanner scanner = conn.createScanner(sConfig.getAccumuloTable(), Authorizations.EMPTY);
-      scanner.setRange(SpanUtil.toRange(span));
-      for (Column col : columns) {
+      scanner.setRange(SpanUtil.toRange(scanConfig.getSpan()));
+      for (Column col : scanConfig.getColumns()) {
         if (col.isQualifierSet()) {
           scanner
               .fetchColumn(ByteUtil.toText(col.getFamily()), ByteUtil.toText(col.getQualifier()));
diff --git a/modules/core/pom.xml b/modules/core/pom.xml
index 51d02ecf..15d1b720 100644
--- a/modules/core/pom.xml
+++ b/modules/core/pom.xml
@@ -41,6 +41,11 @@
       <groupId>io.dropwizard.metrics</groupId>
       <artifactId>metrics-graphite</artifactId>
     </dependency>
+    <dependency>
+      <groupId>log4j</groupId>
+      <artifactId>log4j</artifactId>
+      <version>1.2.16</version>
+    </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-core</artifactId>
@@ -77,10 +82,6 @@
       <groupId>org.apache.zookeeper</groupId>
       <artifactId>zookeeper</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.hdrhistogram</groupId>
-      <artifactId>HdrHistogram</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.mpierce.metrics.reservoir</groupId>
       <artifactId>hdrhistogram-metrics-reservoir</artifactId>
@@ -94,10 +95,5 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-log4j12</artifactId>
-      <scope>test</scope>
-    </dependency>
   </dependencies>
 </project>
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/ColumnIteratorImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/ColumnIteratorImpl.java
new file mode 100644
index 00000000..266df690
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/ColumnIteratorImpl.java
@@ -0,0 +1,90 @@
+/*
+ * 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.fluo.core.impl;
+
+import java.util.Iterator;
+import java.util.Map.Entry;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.core.util.ByteUtil;
+
+/**
+ * Implementation of Column Iterator
+ */
+public class ColumnIteratorImpl implements ColumnIterator {
+
+  private Iterator<Entry<Key, Value>> scanner;
+  private Entry<Key, Value> firstEntry;
+
+  ColumnIteratorImpl(Iterator<Entry<Key, Value>> scanner) {
+    this(null, scanner);
+  }
+
+  ColumnIteratorImpl(Entry<Key, Value> firstEntry, Iterator<Entry<Key, Value>> cols) {
+    this.firstEntry = firstEntry;
+    this.scanner = cols;
+  }
+
+  @Override
+  public boolean hasNext() {
+    return firstEntry != null || scanner.hasNext();
+  }
+
+  // TODO create custom class to return instead of entry
+  @Override
+  public Entry<Column, Bytes> next() {
+    Entry<Key, Value> entry;
+    if (firstEntry != null) {
+      entry = firstEntry;
+      firstEntry = null;
+    } else {
+      entry = scanner.next();
+    }
+    final Bytes cf = ByteUtil.toBytes(entry.getKey().getColumnFamilyData());
+    final Bytes cq = ByteUtil.toBytes(entry.getKey().getColumnQualifierData());
+    final Bytes cv = ByteUtil.toBytes(entry.getKey().getColumnVisibilityData());
+
+    final Column col = new Column(cf, cq, cv);
+    final Bytes val = Bytes.of(entry.getValue().get());
+
+    return new Entry<Column, Bytes>() {
+
+      @Override
+      public Bytes setValue(Bytes value) {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public Bytes getValue() {
+        return val;
+      }
+
+      @Override
+      public Column getKey() {
+        return col;
+      }
+    };
+  }
+
+  @Override
+  public void remove() {
+    scanner.remove();
+  }
+}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/RowIteratorImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/RowIteratorImpl.java
new file mode 100644
index 00000000..efdefebe
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/RowIteratorImpl.java
@@ -0,0 +1,77 @@
+/*
+ * 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.fluo.core.impl;
+
+import java.util.Iterator;
+import java.util.Map.Entry;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
+
+/**
+ * Implementation of RowIterator
+ */
+public class RowIteratorImpl implements RowIterator {
+
+  private final org.apache.accumulo.core.client.RowIterator rowIter;
+
+  RowIteratorImpl(Iterator<Entry<Key, Value>> scanner) {
+    rowIter = new org.apache.accumulo.core.client.RowIterator(scanner);
+  }
+
+  @Override
+  public boolean hasNext() {
+    return rowIter.hasNext();
+  }
+
+  // TODO create custom class to return instead of entry
+  @Override
+  public Entry<Bytes, ColumnIterator> next() {
+    Iterator<Entry<Key, Value>> cols = rowIter.next();
+
+    Entry<Key, Value> entry = cols.next();
+
+    final Bytes row = Bytes.of(entry.getKey().getRowData().toArray());
+    final ColumnIterator coliter = new ColumnIteratorImpl(entry, cols);
+
+    return new Entry<Bytes, ColumnIterator>() {
+
+      @Override
+      public Bytes getKey() {
+        return row;
+      }
+
+      @Override
+      public ColumnIterator getValue() {
+        return coliter;
+      }
+
+      @Override
+      public ColumnIterator setValue(ColumnIterator value) {
+        throw new UnsupportedOperationException();
+      }
+    };
+
+  }
+
+  @Override
+  public void remove() {
+    rowIter.remove();
+  }
+}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/SnapshotScanner.java b/modules/core/src/main/java/org/apache/fluo/core/impl/SnapshotScanner.java
index cd2e008a..2b9959df 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/SnapshotScanner.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/SnapshotScanner.java
@@ -16,13 +16,12 @@
 package org.apache.fluo.core.impl;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.NoSuchElementException;
+import java.util.Set;
 
-import com.google.common.collect.ImmutableSet;
 import org.apache.accumulo.core.client.IteratorSetting;
 import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.client.ScannerBase;
@@ -31,6 +30,7 @@ import org.apache.accumulo.core.data.Key;
 import org.apache.accumulo.core.data.Value;
 import org.apache.fluo.accumulo.iterators.SnapshotIterator;
 import org.apache.fluo.accumulo.util.ColumnConstants;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.api.data.Span;
@@ -41,41 +41,45 @@ import org.apache.fluo.core.util.UtilWaitThread;
 /**
  * Allows users to iterate over entries of a {@link org.apache.fluo.api.client.Snapshot}
  */
-public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
-
-  /**
-   * Immutable options for a SnapshotScanner
-   */
-  public static final class Opts {
-    private final Span span;
-    private final Collection<Column> columns;
-
-    public Opts(Span span, Collection<Column> columns) {
-      this.span = span;
-      this.columns = ImmutableSet.copyOf(columns);
-    }
-
-    public Span getSpan() {
-      return span;
-    }
-
-    public Collection<Column> getColumns() {
-      return columns;
-    }
-  }
+public class SnapshotScanner implements Iterator<Entry<Key, Value>> {
 
   private final long startTs;
   private final Environment env;
   private final TxStats stats;
-  private final Opts config;
+
+  private Iterator<Entry<Key, Value>> iterator;
+  private Entry<Key, Value> next;
+  private ScannerConfiguration config;
 
   static final long INITIAL_WAIT_TIME = 50;
   // TODO make configurable
   static final long MAX_WAIT_TIME = 60000;
 
+  public SnapshotScanner(Environment env, ScannerConfiguration config, long startTs, TxStats stats) {
+    this.env = env;
+    this.config = config;
+    this.startTs = startTs;
+    this.stats = stats;
+    setUpIterator();
+  }
 
+  private void setUpIterator() {
+    Scanner scanner;
+    try {
+      scanner = env.getConnector().createScanner(env.getTable(), env.getAuthorizations());
+    } catch (TableNotFoundException e) {
+      throw new RuntimeException(e);
+    }
+    scanner.clearColumns();
+    scanner.clearScanIterators();
+    scanner.setRange(SpanUtil.toRange(config.getSpan()));
+
+    setupScanner(scanner, config.getColumns(), startTs);
 
-  static void setupScanner(ScannerBase scanner, Collection<Column> columns, long startTs) {
+    this.iterator = scanner.iterator();
+  }
+
+  static void setupScanner(ScannerBase scanner, Set<Column> columns, long startTs) {
     for (Column col : columns) {
       if (col.isQualifierSet()) {
         scanner.fetchColumn(ByteUtil.toText(col.getFamily()), ByteUtil.toText(col.getQualifier()));
@@ -89,153 +93,119 @@ public class SnapshotScanner implements Iterable<Entry<Key, Value>> {
     scanner.addScanIterator(iterConf);
   }
 
-  private class SnapIter implements Iterator<Entry<Key, Value>> {
-
-    private Iterator<Entry<Key, Value>> iterator;
-    private Entry<Key, Value> next;
-    private Opts snapIterConfig;
-
-    SnapIter(Opts config) {
-      this.snapIterConfig = config;
-      setUpIterator();
-    }
-
-    private void setUpIterator() {
-      Scanner scanner;
-      try {
-        scanner = env.getConnector().createScanner(env.getTable(), env.getAuthorizations());
-      } catch (TableNotFoundException e) {
-        throw new RuntimeException(e);
-      }
-      scanner.clearColumns();
-      scanner.clearScanIterators();
-      scanner.setRange(SpanUtil.toRange(snapIterConfig.getSpan()));
-
-      setupScanner(scanner, snapIterConfig.getColumns(), startTs);
-
-      this.iterator = scanner.iterator();
+  @Override
+  public boolean hasNext() {
+    if (next == null) {
+      next = getNext();
     }
 
-    @Override
-    public boolean hasNext() {
-      if (next == null) {
-        next = getNext();
-      }
+    return next != null;
+  }
 
-      return next != null;
+  @Override
+  public Entry<Key, Value> next() {
+    if (!hasNext()) {
+      throw new NoSuchElementException();
     }
 
-    @Override
-    public Entry<Key, Value> next() {
-      if (!hasNext()) {
-        throw new NoSuchElementException();
-      }
+    Entry<Key, Value> tmp = next;
+    next = null;
+    return tmp;
+  }
 
-      Entry<Key, Value> tmp = next;
-      next = null;
-      return tmp;
+  private void resetScanner(Span span) {
+    try {
+      config = (ScannerConfiguration) config.clone();
+    } catch (CloneNotSupportedException e) {
+      throw new RuntimeException(e);
     }
 
-    private void resetScanner(Span span) {
-      snapIterConfig = new Opts(span, snapIterConfig.columns);
-      setUpIterator();
-    }
+    config.setSpan(span);
+    setUpIterator();
+  }
 
-    public void resolveLock(Entry<Key, Value> lockEntry) {
+  public void resolveLock(Entry<Key, Value> lockEntry) {
 
-      // read ahead a little bit looking for other locks to resolve
+    // read ahead a little bit looking for other locks to resolve
 
-      long startTime = System.currentTimeMillis();
-      long waitTime = INITIAL_WAIT_TIME;
+    long startTime = System.currentTimeMillis();
+    long waitTime = INITIAL_WAIT_TIME;
 
-      List<Entry<Key, Value>> locks = new ArrayList<>();
-      locks.add(lockEntry);
-      int amountRead = 0;
-      int numRead = 0;
+    List<Entry<Key, Value>> locks = new ArrayList<>();
+    locks.add(lockEntry);
+    int amountRead = 0;
+    int numRead = 0;
 
-      RowColumn origEnd = snapIterConfig.getSpan().getEnd();
-      boolean isEndInclusive = snapIterConfig.getSpan().isEndInclusive();
+    RowColumn origEnd = config.getSpan().getEnd();
+    boolean isEndInclusive = config.getSpan().isEndInclusive();
 
-      while (true) {
-        while (iterator.hasNext()) {
-          Entry<Key, Value> entry = iterator.next();
+    while (true) {
+      while (iterator.hasNext()) {
+        Entry<Key, Value> entry = iterator.next();
 
-          long colType = entry.getKey().getTimestamp() & ColumnConstants.PREFIX_MASK;
+        long colType = entry.getKey().getTimestamp() & ColumnConstants.PREFIX_MASK;
 
-          if (colType == ColumnConstants.LOCK_PREFIX) {
-            locks.add(entry);
-          }
+        if (colType == ColumnConstants.LOCK_PREFIX) {
+          locks.add(entry);
+        }
 
-          amountRead += entry.getKey().getSize() + entry.getValue().getSize();
-          numRead++;
+        amountRead += entry.getKey().getSize() + entry.getValue().getSize();
+        numRead++;
 
-          if (numRead > 100 || amountRead > 1 << 12) {
-            break;
-          }
+        if (numRead > 100 || amountRead > 1 << 12) {
+          break;
         }
+      }
 
-        boolean resolvedLocks = LockResolver.resolveLocks(env, startTs, stats, locks, startTime);
+      boolean resolvedLocks = LockResolver.resolveLocks(env, startTs, stats, locks, startTime);
 
-        if (!resolvedLocks) {
-          UtilWaitThread.sleep(waitTime);
-          stats.incrementLockWaitTime(waitTime);
-          waitTime = Math.min(MAX_WAIT_TIME, waitTime * 2);
+      if (!resolvedLocks) {
+        UtilWaitThread.sleep(waitTime);
+        stats.incrementLockWaitTime(waitTime);
+        waitTime = Math.min(MAX_WAIT_TIME, waitTime * 2);
 
-          RowColumn start = SpanUtil.toRowColumn(locks.get(0).getKey());
-          RowColumn end = SpanUtil.toRowColumn(locks.get(locks.size() - 1).getKey()).following();
+        RowColumn start = SpanUtil.toRowColumn(locks.get(0).getKey());
+        RowColumn end = SpanUtil.toRowColumn(locks.get(locks.size() - 1).getKey()).following();
 
-          resetScanner(new Span(start, true, end, false));
+        resetScanner(new Span(start, true, end, false));
 
-          locks.clear();
+        locks.clear();
 
-        } else {
-          break;
-        }
+      } else {
+        break;
       }
+    }
 
-      RowColumn start = SpanUtil.toRowColumn(lockEntry.getKey());
+    RowColumn start = SpanUtil.toRowColumn(lockEntry.getKey());
 
-      resetScanner(new Span(start, true, origEnd, isEndInclusive));
-    }
+    resetScanner(new Span(start, true, origEnd, isEndInclusive));
+  }
 
-    public Entry<Key, Value> getNext() {
-      mloop: while (true) {
-        // its possible a next could exist then be rolled back
-        if (!iterator.hasNext()) {
-          return null;
-        }
+  public Entry<Key, Value> getNext() {
+    mloop: while (true) {
+      // its possible a next could exist then be rolled back
+      if (!iterator.hasNext()) {
+        return null;
+      }
 
-        Entry<Key, Value> entry = iterator.next();
+      Entry<Key, Value> entry = iterator.next();
 
-        long colType = entry.getKey().getTimestamp() & ColumnConstants.PREFIX_MASK;
+      long colType = entry.getKey().getTimestamp() & ColumnConstants.PREFIX_MASK;
 
-        if (colType == ColumnConstants.LOCK_PREFIX) {
-          resolveLock(entry);
-          continue mloop;
-        } else if (colType == ColumnConstants.DATA_PREFIX) {
-          stats.incrementEntriesReturned(1);
-          return entry;
-        } else {
-          throw new IllegalArgumentException("Unexpected column type " + colType);
-        }
+      if (colType == ColumnConstants.LOCK_PREFIX) {
+        resolveLock(entry);
+        continue mloop;
+      } else if (colType == ColumnConstants.DATA_PREFIX) {
+        stats.incrementEntriesReturned(1);
+        return entry;
+      } else {
+        throw new IllegalArgumentException("Unexpected column type " + colType);
       }
     }
-
-    @Override
-    public void remove() {
-      iterator.remove();
-    }
-  }
-
-  SnapshotScanner(Environment env, Opts config, long startTs, TxStats stats) {
-    this.env = env;
-    this.config = config;
-    this.startTs = startTs;
-    this.stats = stats;
   }
 
   @Override
-  public Iterator<Entry<Key, Value>> iterator() {
-    return new SnapIter(config);
+  public void remove() {
+    iterator.remove();
   }
 }
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
index c4c429ea..3b0b85ce 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/impl/TransactionImpl.java
@@ -29,7 +29,6 @@ import java.util.Set;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Iterables;
 import com.google.common.collect.Iterators;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
@@ -54,23 +53,22 @@ import org.apache.fluo.accumulo.util.ColumnConstants;
 import org.apache.fluo.accumulo.values.DelLockValue;
 import org.apache.fluo.accumulo.values.LockValue;
 import org.apache.fluo.api.client.Snapshot;
-import org.apache.fluo.api.client.scanner.ScannerBuilder;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.ColumnValue;
 import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.api.data.Span;
 import org.apache.fluo.api.exceptions.AlreadySetException;
 import org.apache.fluo.api.exceptions.CommitException;
 import org.apache.fluo.api.exceptions.FluoException;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.core.async.AsyncCommitObserver;
 import org.apache.fluo.core.async.AsyncConditionalWriter;
 import org.apache.fluo.core.async.AsyncTransaction;
 import org.apache.fluo.core.async.SyncCommitObserver;
 import org.apache.fluo.core.exceptions.AlreadyAcknowledgedException;
 import org.apache.fluo.core.exceptions.StaleScanException;
-import org.apache.fluo.core.impl.scanner.ColumnScannerImpl;
-import org.apache.fluo.core.impl.scanner.ScannerBuilderImpl;
 import org.apache.fluo.core.oracle.Stamp;
 import org.apache.fluo.core.util.ColumnUtil;
 import org.apache.fluo.core.util.ConditionalFlutation;
@@ -209,46 +207,36 @@ public class TransactionImpl implements AsyncTransaction, Snapshot {
     return ret;
   }
 
+  @Override
+  public RowIterator get(ScannerConfiguration config) {
+    checkIfOpen();
+    return getImpl(config);
+  }
+
   private Map<Column, Bytes> getImpl(Bytes row, Set<Column> columns) {
 
     // TODO push visibility filtering to server side?
 
     env.getSharedResources().getVisCache().validate(columns);
 
-    boolean shouldCopy = false;
-
+    ScannerConfiguration config = new ScannerConfiguration();
+    config.setSpan(Span.exact(row));
     for (Column column : columns) {
-      if (column.isVisibilitySet()) {
-        shouldCopy = true;
-      }
+      config.fetchColumn(column.getFamily(), column.getQualifier());
     }
 
-    SnapshotScanner.Opts opts;
-    if (shouldCopy) {
-      HashSet<Column> cols = new HashSet<Column>();
-      for (Column column : columns) {
-        if (column.isVisibilitySet()) {
-          cols.add(new Column(column.getFamily(), column.getQualifier()));
-        } else {
-          cols.add(column);
-        }
-      }
-      opts = new SnapshotScanner.Opts(Span.exact(row), columns);
-    } else {
-      opts = new SnapshotScanner.Opts(Span.exact(row), columns);
-    }
+    RowIterator iter = getImpl(config);
 
     Map<Column, Bytes> ret = new HashMap<>();
 
-    Iterable<ColumnValue> scanner =
-        Iterables.transform(new SnapshotScanner(env, opts, startTs, stats), ColumnScannerImpl.E2CV);
-    for (ColumnValue cv : scanner) {
-      if (shouldCopy) {
-        if (columns.contains(cv.getColumn())) {
-          ret.put(cv.getColumn(), cv.getValue());
+    while (iter.hasNext()) {
+      Entry<Bytes, ColumnIterator> entry = iter.next();
+      ColumnIterator citer = entry.getValue();
+      while (citer.hasNext()) {
+        Entry<Column, Bytes> centry = citer.next();
+        if (columns.contains(centry.getKey())) {
+          ret.put(centry.getKey(), centry.getValue());
         }
-      } else {
-        ret.put(cv.getColumn(), cv.getValue());
       }
     }
 
@@ -258,10 +246,8 @@ public class TransactionImpl implements AsyncTransaction, Snapshot {
     return ret;
   }
 
-  @Override
-  public ScannerBuilder scanner() {
-    checkIfOpen();
-    return new ScannerBuilderImpl(this);
+  private RowIterator getImpl(ScannerConfiguration config) {
+    return new RowIteratorImpl(new SnapshotScanner(this.env, config, startTs, stats));
   }
 
   private void updateColumnsRead(Bytes row, Set<Column> columns) {
@@ -1207,7 +1193,4 @@ public class TransactionImpl implements AsyncTransaction, Snapshot {
     cd.commitObserver.committed();
   }
 
-  public SnapshotScanner newSnapshotScanner(Span span, Collection<Column> columns) {
-    return new SnapshotScanner(env, new SnapshotScanner.Opts(span, columns), startTs, stats);
-  }
 }
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/CellScannerImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/CellScannerImpl.java
deleted file mode 100644
index 06e68d6c..00000000
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/CellScannerImpl.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.fluo.core.impl.scanner;
-
-import java.util.Iterator;
-import java.util.Map.Entry;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Iterators;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.Value;
-import org.apache.fluo.api.client.scanner.CellScanner;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumnValue;
-import org.apache.fluo.core.util.ByteUtil;
-
-public class CellScannerImpl implements CellScanner {
-
-  private Iterable<Entry<Key, Value>> snapshot;
-
-  private static final Function<Entry<Key, Value>, RowColumnValue> E2RCV =
-      new Function<Entry<Key, Value>, RowColumnValue>() {
-        @Override
-        public RowColumnValue apply(Entry<Key, Value> entry) {
-          Bytes row = ByteUtil.toBytes(entry.getKey().getRowData());
-          Bytes cf = ByteUtil.toBytes(entry.getKey().getColumnFamilyData());
-          Bytes cq = ByteUtil.toBytes(entry.getKey().getColumnQualifierData());
-          Bytes cv = ByteUtil.toBytes(entry.getKey().getColumnVisibilityData());
-          Column col = new Column(cf, cq, cv);
-          Bytes val = Bytes.of(entry.getValue().get());
-          return new RowColumnValue(row, col, val);
-        }
-      };
-
-  CellScannerImpl(Iterable<Entry<Key, Value>> snapshot) {
-    this.snapshot = snapshot;
-  }
-
-  @Override
-  public Iterator<RowColumnValue> iterator() {
-    return Iterators.transform(snapshot.iterator(), E2RCV);
-  }
-}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/ColumnScannerImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/ColumnScannerImpl.java
deleted file mode 100644
index 7cde4a5e..00000000
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/ColumnScannerImpl.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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.fluo.core.impl.scanner;
-
-import java.util.Iterator;
-import java.util.Map.Entry;
-
-import com.google.common.base.Function;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Iterators;
-import com.google.common.collect.PeekingIterator;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.Value;
-import org.apache.fluo.api.client.scanner.ColumnScanner;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.ColumnValue;
-import org.apache.fluo.core.util.ByteUtil;
-
-public class ColumnScannerImpl implements ColumnScanner {
-
-  public static final Function<Entry<Key, Value>, ColumnValue> E2CV =
-      new Function<Entry<Key, Value>, ColumnValue>() {
-        @Override
-        public ColumnValue apply(Entry<Key, Value> entry) {
-          Bytes cf = ByteUtil.toBytes(entry.getKey().getColumnFamilyData());
-          Bytes cq = ByteUtil.toBytes(entry.getKey().getColumnQualifierData());
-          Bytes cv = ByteUtil.toBytes(entry.getKey().getColumnVisibilityData());
-          Column col = new Column(cf, cq, cv);
-          Bytes val = Bytes.of(entry.getValue().get());
-          return new ColumnValue(col, val);
-        }
-      };
-
-  private PeekingIterator<Entry<Key, Value>> peekingIter;
-  private Bytes row;
-  private Iterator<ColumnValue> iter;
-  private boolean gotIter = false;
-
-  ColumnScannerImpl(Iterator<Entry<Key, Value>> e) {
-    peekingIter = Iterators.peekingIterator(e);
-    row = ByteUtil.toBytes(peekingIter.peek().getKey().getRowData());
-    iter = Iterators.transform(peekingIter, E2CV);
-  }
-
-  @Override
-  public Iterator<ColumnValue> iterator() {
-    Preconditions.checkState(!gotIter,
-        "Unfortunately this implementation only support getting the iterator once");
-    gotIter = true;
-    return iter;
-  }
-
-  @Override
-  public Bytes getRow() {
-    return row;
-  }
-
-  @Override
-  public String getsRow() {
-    return getRow().toString();
-  }
-}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/RowScannerImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/RowScannerImpl.java
deleted file mode 100644
index 4a9eb38c..00000000
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/RowScannerImpl.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.fluo.core.impl.scanner;
-
-import java.util.Iterator;
-import java.util.Map.Entry;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Iterators;
-import org.apache.accumulo.core.client.RowIterator;
-import org.apache.accumulo.core.data.Key;
-import org.apache.accumulo.core.data.Value;
-import org.apache.fluo.api.client.scanner.ColumnScanner;
-import org.apache.fluo.api.client.scanner.RowScanner;
-
-public class RowScannerImpl implements RowScanner {
-
-  private Iterable<Entry<Key, Value>> snapshot;
-
-  private static final Function<Iterator<Entry<Key, Value>>, ColumnScanner> RI2CS =
-      new Function<Iterator<Entry<Key, Value>>, ColumnScanner>() {
-        @Override
-        public ColumnScanner apply(Iterator<Entry<Key, Value>> input) {
-          return new ColumnScannerImpl(input);
-        }
-      };
-
-  RowScannerImpl(Iterable<Entry<Key, Value>> snapshot) {
-    this.snapshot = snapshot;
-  }
-
-  @Override
-  public Iterator<ColumnScanner> iterator() {
-    RowIterator rowiter = new RowIterator(snapshot.iterator());
-    return Iterators.transform(rowiter, RI2CS);
-  }
-}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/ScannerBuilderImpl.java b/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/ScannerBuilderImpl.java
deleted file mode 100644
index 8c833d55..00000000
--- a/modules/core/src/main/java/org/apache/fluo/core/impl/scanner/ScannerBuilderImpl.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.fluo.core.impl.scanner;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Objects;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableSet;
-import org.apache.fluo.api.client.scanner.CellScanner;
-import org.apache.fluo.api.client.scanner.RowScanner;
-import org.apache.fluo.api.client.scanner.RowScannerBuilder;
-import org.apache.fluo.api.client.scanner.ScannerBuilder;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.Span;
-import org.apache.fluo.core.impl.SnapshotScanner;
-import org.apache.fluo.core.impl.TransactionImpl;
-
-public class ScannerBuilderImpl implements ScannerBuilder {
-
-  private static final Span EMPTY_SPAN = new Span();
-
-  private TransactionImpl tx;
-  private Span span = EMPTY_SPAN;
-  private Collection<Column> columns = Collections.emptyList();
-
-  public ScannerBuilderImpl(TransactionImpl tx) {
-    this.tx = tx;
-  }
-
-  @Override
-  public ScannerBuilder over(Span span) {
-    Objects.requireNonNull(span);
-    this.span = span;
-    return this;
-  }
-
-  private void setColumns(Collection<Column> columns) {
-    for (Column column : columns) {
-      Preconditions.checkArgument(!column.isVisibilitySet(),
-          "Fetching visibility is not currently supported");
-    }
-    this.columns = columns;
-  }
-
-  @Override
-  public ScannerBuilder fetch(Collection<Column> columns) {
-    Objects.requireNonNull(columns);
-    setColumns(ImmutableSet.copyOf(columns));
-    return this;
-  }
-
-  @Override
-  public ScannerBuilder fetch(Column... columns) {
-    Objects.requireNonNull(columns);
-    setColumns(ImmutableSet.copyOf(columns));
-    return this;
-  }
-
-  @Override
-  public CellScanner build() {
-    SnapshotScanner snapScanner = tx.newSnapshotScanner(span, columns);
-    return new CellScannerImpl(snapScanner);
-  }
-
-  @Override
-  public RowScannerBuilder byRow() {
-    return new RowScannerBuilder() {
-      @Override
-      public RowScanner build() {
-        SnapshotScanner snapScanner = tx.newSnapshotScanner(span, columns);
-        return new RowScannerImpl(snapScanner);
-      }
-    };
-  }
-}
diff --git a/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java b/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
index fe5c21d1..da6b8dd4 100644
--- a/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
+++ b/modules/core/src/main/java/org/apache/fluo/core/log/TracingTransaction.java
@@ -23,13 +23,14 @@ import java.util.Set;
 import com.google.common.base.Function;
 import com.google.common.collect.Iterators;
 import org.apache.fluo.api.client.Snapshot;
-import org.apache.fluo.api.client.scanner.ScannerBuilder;
 import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.api.exceptions.AlreadySetException;
 import org.apache.fluo.api.exceptions.CommitException;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.core.async.AsyncCommitObserver;
 import org.apache.fluo.core.async.AsyncTransaction;
 import org.apache.fluo.core.impl.Notification;
@@ -178,10 +179,10 @@ public class TracingTransaction implements AsyncTransaction, Snapshot {
   }
 
   @Override
-  public ScannerBuilder scanner() {
+  public RowIterator get(ScannerConfiguration config) {
     // TODO log something better (see fluo-425)
-    log.trace("txid: {} newScanner()", txid);
-    return tx.scanner();
+    log.trace("txid: {} get(ScannerConfiguration)", txid);
+    return tx.get(config);
   }
 
   @Override
diff --git a/modules/distribution/pom.xml b/modules/distribution/pom.xml
index 7d4ea20b..0727749b 100644
--- a/modules/distribution/pom.xml
+++ b/modules/distribution/pom.xml
@@ -52,14 +52,6 @@
       <groupId>org.apache.fluo</groupId>
       <artifactId>fluo-mini</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>log4j-over-slf4j</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-log4j12</artifactId>
-    </dependency>
   </dependencies>
   <build>
     <plugins>
diff --git a/modules/integration/pom.xml b/modules/integration/pom.xml
index e989afd9..61cffa95 100644
--- a/modules/integration/pom.xml
+++ b/modules/integration/pom.xml
@@ -26,8 +26,8 @@
   <description>This module contains Fluo integration tests</description>
   <dependencies>
     <dependency>
-      <groupId>commons-configuration</groupId>
-      <artifactId>commons-configuration</artifactId>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -40,6 +40,12 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>log4j</groupId>
+      <artifactId>log4j</artifactId>
+      <version>1.2.16</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-core</artifactId>
@@ -86,13 +92,8 @@
       <scope>test</scope>
     </dependency>
     <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-api</artifactId>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.slf4j</groupId>
-      <artifactId>slf4j-log4j12</artifactId>
+      <groupId>org.apache.zookeeper</groupId>
+      <artifactId>zookeeper</artifactId>
       <scope>test</scope>
     </dependency>
   </dependencies>
diff --git a/modules/mapreduce/pom.xml b/modules/mapreduce/pom.xml
index 2bd2b358..0eed66bf 100644
--- a/modules/mapreduce/pom.xml
+++ b/modules/mapreduce/pom.xml
@@ -25,14 +25,6 @@
   <name>Fluo MapReduce</name>
   <description>This module provides utility code for MapReduce jobs that read from or write to a Fluo table.</description>
   <dependencies>
-    <dependency>
-      <groupId>com.google.guava</groupId>
-      <artifactId>guava</artifactId>
-    </dependency>
-    <dependency>
-      <groupId>commons-configuration</groupId>
-      <artifactId>commons-configuration</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-core</artifactId>
@@ -70,6 +62,28 @@
       <scope>test</scope>
     </dependency>
   </dependencies>
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-dependency-plugin</artifactId>
+        <version>2.10</version>
+        <executions>
+          <execution>
+            <id>analyze</id>
+            <goals>
+              <goal>analyze-only</goal>
+            </goals>
+            <configuration>
+              <ignoredDependencies combine.children="append">
+                <ignoredDependency>org.apache.accumulo:accumulo-minicluster:jar:${accumulo.version}</ignoredDependency>
+              </ignoredDependencies>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
   <profiles>
     <profile>
       <id>mini-accumulo</id>
diff --git a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoEntryInputFormat.java b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoEntryInputFormat.java
index 18c091c6..bdea2993 100644
--- a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoEntryInputFormat.java
+++ b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoEntryInputFormat.java
@@ -19,20 +19,21 @@ import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.HashSet;
-import java.util.Iterator;
 import java.util.List;
+import java.util.Map.Entry;
 
 import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
 import org.apache.accumulo.core.client.mapreduce.RangeInputSplit;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
 import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.config.SimpleConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.api.data.RowColumn;
-import org.apache.fluo.api.data.RowColumnValue;
 import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.core.impl.Environment;
 import org.apache.fluo.core.impl.TransactionImpl;
 import org.apache.fluo.core.util.SpanUtil;
@@ -58,10 +59,13 @@ public class FluoEntryInputFormat extends InputFormat<RowColumn, Bytes> {
 
     return new RecordReader<RowColumn, Bytes>() {
 
-      private RowColumnValue rowColVal;
+      private RowColumn rowCol;
+      private Bytes val;
+      private RowIterator rowIter;
+      private Bytes row;
+      private ColumnIterator colIter = null;
       private Environment env = null;
       private TransactionImpl ti = null;
-      private Iterator<RowColumnValue> cellIterator;
 
       @Override
       public void close() throws IOException {
@@ -76,12 +80,12 @@ public class FluoEntryInputFormat extends InputFormat<RowColumn, Bytes> {
 
       @Override
       public RowColumn getCurrentKey() throws IOException, InterruptedException {
-        return rowColVal;
+        return rowCol;
       }
 
       @Override
       public Bytes getCurrentValue() throws IOException, InterruptedException {
-        return rowColVal.getValue();
+        return val;
       }
 
       @Override
@@ -105,14 +109,13 @@ public class FluoEntryInputFormat extends InputFormat<RowColumn, Bytes> {
           // TODO this uses non public Accumulo API!
           RangeInputSplit ris = (RangeInputSplit) split;
           Span span = SpanUtil.toSpan(ris.getRange());
-
-          HashSet<Column> columns = new HashSet<>();
+          ScannerConfiguration sc = new ScannerConfiguration().setSpan(span);
 
           for (String fam : context.getConfiguration().getStrings(FAMS_CONF_KEY, new String[0])) {
-            columns.add(new Column(fam));
+            sc.fetchColumnFamily(Bytes.of(fam));
           }
 
-          cellIterator = ti.scanner().over(span).fetch(columns).build().iterator();
+          rowIter = ti.get(sc);
         } catch (Exception e) {
           throw new IOException(e);
         }
@@ -120,12 +123,19 @@ public class FluoEntryInputFormat extends InputFormat<RowColumn, Bytes> {
 
       @Override
       public boolean nextKeyValue() throws IOException, InterruptedException {
-        if (cellIterator.hasNext()) {
-          rowColVal = cellIterator.next();
-          return true;
-        } else {
-          rowColVal = null;
-          return false;
+        while (true) {
+          if ((colIter != null) && (colIter.hasNext())) {
+            Entry<Column, Bytes> colEntry = colIter.next();
+            rowCol = new RowColumn(row, colEntry.getKey());
+            val = colEntry.getValue();
+            return true;
+          } else if (rowIter.hasNext()) {
+            Entry<Bytes, ColumnIterator> rowEntry = rowIter.next();
+            row = rowEntry.getKey();
+            colIter = rowEntry.getValue();
+          } else {
+            return false;
+          }
         }
       }
     };
@@ -168,6 +178,7 @@ public class FluoEntryInputFormat extends InputFormat<RowColumn, Bytes> {
     }
   }
 
+  // TODO support text
   public static void fetchFamilies(Job job, String... fams) {
     job.getConfiguration().setStrings(FAMS_CONF_KEY, fams);
   }
diff --git a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoRowInputFormat.java b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoRowInputFormat.java
index 9dc5e727..c32b9fc2 100644
--- a/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoRowInputFormat.java
+++ b/modules/mapreduce/src/main/java/org/apache/fluo/mapreduce/FluoRowInputFormat.java
@@ -19,20 +19,19 @@ import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.HashSet;
-import java.util.Iterator;
 import java.util.List;
+import java.util.Map.Entry;
 
 import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
 import org.apache.accumulo.core.client.mapreduce.RangeInputSplit;
 import org.apache.accumulo.core.client.security.tokens.PasswordToken;
-import org.apache.fluo.api.client.scanner.ColumnScanner;
 import org.apache.fluo.api.config.FluoConfiguration;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.config.SimpleConfiguration;
 import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.ColumnValue;
 import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.core.impl.Environment;
 import org.apache.fluo.core.impl.TransactionImpl;
 import org.apache.fluo.core.util.SpanUtil;
@@ -46,20 +45,20 @@ import org.apache.hadoop.mapreduce.TaskAttemptContext;
 /**
  * This input format reads a consistent snapshot of Fluo rows from a Fluo table.
  */
-public class FluoRowInputFormat extends InputFormat<Bytes, Iterator<ColumnValue>> {
+public class FluoRowInputFormat extends InputFormat<Bytes, ColumnIterator> {
 
   private static String TIMESTAMP_CONF_KEY = FluoRowInputFormat.class.getName() + ".timestamp";
   private static String PROPS_CONF_KEY = FluoRowInputFormat.class.getName() + ".props";
   private static String FAMS_CONF_KEY = FluoRowInputFormat.class.getName() + ".families";
 
   @Override
-  public RecordReader<Bytes, Iterator<ColumnValue>> createRecordReader(InputSplit split,
+  public RecordReader<Bytes, ColumnIterator> createRecordReader(InputSplit split,
       TaskAttemptContext context) throws IOException, InterruptedException {
 
-    return new RecordReader<Bytes, Iterator<ColumnValue>>() {
+    return new RecordReader<Bytes, ColumnIterator>() {
 
-      private ColumnScanner colScanner;
-      private Iterator<ColumnScanner> rowIterator;
+      private Entry<Bytes, ColumnIterator> entry;
+      private RowIterator rowIter;
       private Environment env = null;
       private TransactionImpl ti = null;
 
@@ -76,12 +75,12 @@ public class FluoRowInputFormat extends InputFormat<Bytes, Iterator<ColumnValue>
 
       @Override
       public Bytes getCurrentKey() throws IOException, InterruptedException {
-        return colScanner.getRow();
+        return entry.getKey();
       }
 
       @Override
-      public Iterator<ColumnValue> getCurrentValue() throws IOException, InterruptedException {
-        return colScanner.iterator();
+      public ColumnIterator getCurrentValue() throws IOException, InterruptedException {
+        return entry.getValue();
       }
 
       @Override
@@ -105,15 +104,13 @@ public class FluoRowInputFormat extends InputFormat<Bytes, Iterator<ColumnValue>
           // TODO this uses non public Accumulo API!
           RangeInputSplit ris = (RangeInputSplit) split;
           Span span = SpanUtil.toSpan(ris.getRange());
-
-          HashSet<Column> columns = new HashSet<>();
+          ScannerConfiguration sc = new ScannerConfiguration().setSpan(span);
 
           for (String fam : context.getConfiguration().getStrings(FAMS_CONF_KEY, new String[0])) {
-            columns.add(new Column(fam));
+            sc.fetchColumnFamily(Bytes.of(fam));
           }
 
-          rowIterator = ti.scanner().over(span).fetch(columns).byRow().build().iterator();
-
+          rowIter = ti.get(sc);
         } catch (Exception e) {
           throw new IOException(e);
         }
@@ -121,8 +118,8 @@ public class FluoRowInputFormat extends InputFormat<Bytes, Iterator<ColumnValue>
 
       @Override
       public boolean nextKeyValue() throws IOException, InterruptedException {
-        if (rowIterator.hasNext()) {
-          colScanner = rowIterator.next();
+        if (rowIter.hasNext()) {
+          entry = rowIter.next();
           return true;
         }
         return false;
diff --git a/modules/mini/pom.xml b/modules/mini/pom.xml
index 1d05bef1..b554f354 100644
--- a/modules/mini/pom.xml
+++ b/modules/mini/pom.xml
@@ -30,10 +30,6 @@
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>
     </dependency>
-    <dependency>
-      <groupId>commons-configuration</groupId>
-      <artifactId>commons-configuration</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.accumulo</groupId>
       <artifactId>accumulo-core</artifactId>
@@ -49,10 +45,6 @@
         </exclusion>
       </exclusions>
     </dependency>
-    <dependency>
-      <groupId>org.apache.fluo</groupId>
-      <artifactId>fluo-accumulo</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.apache.fluo</groupId>
       <artifactId>fluo-api</artifactId>
diff --git a/pom.xml b/pom.xml
index 1f46eab4..9c80bbd6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -330,6 +330,29 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-dependency-plugin</artifactId>
+        <version>2.10</version>
+        <executions>
+          <execution>
+            <id>analyze</id>
+            <goals>
+              <goal>analyze-only</goal>
+            </goals>
+            <configuration>
+              <failOnWarning>true</failOnWarning>
+              <ignoredDependencies>
+                <ignoredDependency>org.apache.hadoop:hadoop-common:jar:2.6.3</ignoredDependency>
+                <ignoredDependency>org.apache.hadoop:hadoop-client:jar:2.6.3</ignoredDependency>
+                <ignoredDependency>org.apache.hadoop:hadoop-mapreduce-client-core:jar:2.6.3</ignoredDependency>
+                <ignoredDependency>org.apache.hadoop:hadoop-yarn-api:jar:2.6.3</ignoredDependency>
+                <ignoredDependency>org.apache.hadoop:hadoop-yarn-client:jar:2.6.3</ignoredDependency>
+              </ignoredDependencies>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
   <profiles>


Test patch:
diff --git a/modules/cluster/src/test/java/org/apache/fluo/cluster/runner/ScanTest.java b/modules/cluster/src/test/java/org/apache/fluo/cluster/runner/ScanTest.java
index a84e8ead..2d7c17d8 100644
--- a/modules/cluster/src/test/java/org/apache/fluo/cluster/runner/ScanTest.java
+++ b/modules/cluster/src/test/java/org/apache/fluo/cluster/runner/ScanTest.java
@@ -16,10 +16,10 @@
 package org.apache.fluo.cluster.runner;
 
 import com.beust.jcommander.JCommander;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.api.data.Span;
-import org.apache.fluo.core.impl.SnapshotScanner;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -28,16 +28,16 @@ import org.junit.Test;
  */
 public class ScanTest {
 
-  private SnapshotScanner.Opts parseArgs(String args) {
+  private ScannerConfiguration parseArgs(String args) {
     ScanOptions options = new ScanOptions();
     JCommander jcommand = new JCommander(options);
     jcommand.parse(args.split(" "));
-    return new SnapshotScanner.Opts(AppRunner.getSpan(options), AppRunner.getColumns(options));
+    return AppRunner.buildScanConfig(options);
   }
 
   @Test
   public void testValidInput() {
-    SnapshotScanner.Opts config;
+    ScannerConfiguration config;
 
     config = parseArgs("");
     Assert.assertEquals(RowColumn.EMPTY, config.getSpan().getStart());
diff --git a/modules/core/src/test/java/org/apache/fluo/core/config/ScannerConfigurationTest.java b/modules/core/src/test/java/org/apache/fluo/core/config/ScannerConfigurationTest.java
new file mode 100644
index 00000000..f3cc7af1
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/fluo/core/config/ScannerConfigurationTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.fluo.core.config;
+
+import org.apache.fluo.api.config.ScannerConfiguration;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.data.Span;
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * Unit test for ScannerConfiguration class
+ */
+public class ScannerConfigurationTest {
+
+  @Test
+  public void testSetGet() {
+
+    ScannerConfiguration config = new ScannerConfiguration();
+    Assert.assertEquals(new Span(), config.getSpan());
+    Assert.assertEquals(0, config.getColumns().size());
+
+    config = new ScannerConfiguration();
+    config.setSpan(Span.exact("row1"));
+    Assert.assertEquals(Span.exact("row1"), config.getSpan());
+    Assert.assertEquals(0, config.getColumns().size());
+
+    config = new ScannerConfiguration();
+    config.fetchColumnFamily(Bytes.of("cf1"));
+    Assert.assertEquals(1, config.getColumns().size());
+    Assert.assertEquals(new Column("cf1"), config.getColumns().iterator().next());
+
+    config = new ScannerConfiguration();
+    config.fetchColumn(Bytes.of("cf2"), Bytes.of("cq2"));
+    Assert.assertEquals(1, config.getColumns().size());
+    Assert.assertEquals(new Column("cf2", "cq2"), config.getColumns().iterator().next());
+
+    config = new ScannerConfiguration();
+    config.fetchColumnFamily(Bytes.of("a"));
+    config.fetchColumnFamily(Bytes.of("b"));
+    config.fetchColumnFamily(Bytes.of("a"));
+    Assert.assertEquals(2, config.getColumns().size());
+
+    config.clearColumns();
+    Assert.assertEquals(0, config.getColumns().size());
+  }
+
+  @Test
+  public void testNullSet() {
+
+    ScannerConfiguration config = new ScannerConfiguration();
+
+    try {
+      config.setSpan(null);
+      Assert.fail();
+    } catch (NullPointerException e) {
+    }
+
+    try {
+      config.fetchColumnFamily(null);
+      Assert.fail();
+    } catch (NullPointerException e) {
+    }
+
+    try {
+      config.fetchColumn(null, Bytes.of("qual"));
+      Assert.fail();
+    } catch (NullPointerException e) {
+    }
+
+    try {
+      config.fetchColumn(Bytes.of("fam"), null);
+      Assert.fail();
+    } catch (NullPointerException e) {
+    }
+
+  }
+}
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/ITBase.java b/modules/integration/src/test/java/org/apache/fluo/integration/ITBase.java
index 5e2a7f9f..4cdf1462 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/ITBase.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/ITBase.java
@@ -18,6 +18,7 @@ package org.apache.fluo.integration;
 import java.io.File;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map.Entry;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.accumulo.core.client.Connector;
@@ -31,7 +32,11 @@ import org.apache.fluo.api.client.FluoClient;
 import org.apache.fluo.api.client.Snapshot;
 import org.apache.fluo.api.config.FluoConfiguration;
 import org.apache.fluo.api.config.ObserverConfiguration;
-import org.apache.fluo.api.data.RowColumnValue;
+import org.apache.fluo.api.config.ScannerConfiguration;
+import org.apache.fluo.api.data.Bytes;
+import org.apache.fluo.api.data.Column;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 
@@ -94,12 +99,18 @@ public class ITBase {
 
   protected void printSnapshot() throws Exception {
     try (Snapshot s = client.newSnapshot()) {
-      System.out.println("== snapshot start ==");
+      RowIterator iter = s.get(new ScannerConfiguration());
 
-      for (RowColumnValue rcv : s.scanner().build()) {
-        System.out.println(rcv.getRow() + " " + rcv.getColumn() + "\t" + rcv.getValue());
+      System.out.println("== snapshot start ==");
+      while (iter.hasNext()) {
+        Entry<Bytes, ColumnIterator> rowEntry = iter.next();
+        ColumnIterator citer = rowEntry.getValue();
+        while (citer.hasNext()) {
+          Entry<Column, Bytes> colEntry = citer.next();
+          System.out.println(rowEntry.getKey() + " " + colEntry.getKey() + "\t"
+              + colEntry.getValue());
+        }
       }
-
       System.out.println("=== snapshot end ===");
     }
   }
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/ITBaseImpl.java b/modules/integration/src/test/java/org/apache/fluo/integration/ITBaseImpl.java
index 3a541d95..e2ed47aa 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/ITBaseImpl.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/ITBaseImpl.java
@@ -77,7 +77,6 @@ public class ITBaseImpl extends ITBase {
     config.setTransactionRollbackTime(1, TimeUnit.SECONDS);
     config.addObservers(getObservers());
     config.setProperty(FluoConfigurationImpl.ZK_UPDATE_PERIOD_PROP, "1000");
-    config.setMiniStartAccumulo(false);
 
     try (FluoAdmin admin = FluoFactory.newAdmin(config)) {
       InitOpts opts = new InitOpts().setClearZookeeper(true).setClearTable(true);
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/TestTransaction.java b/modules/integration/src/test/java/org/apache/fluo/integration/TestTransaction.java
index ceda1930..b64cda53 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/TestTransaction.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/TestTransaction.java
@@ -31,13 +31,14 @@ import org.apache.fluo.accumulo.iterators.NotificationIterator;
 import org.apache.fluo.accumulo.util.ColumnConstants;
 import org.apache.fluo.accumulo.util.NotificationUtil;
 import org.apache.fluo.api.client.TransactionBase;
-import org.apache.fluo.api.client.scanner.ScannerBuilder;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.api.data.Span;
 import org.apache.fluo.api.exceptions.AlreadySetException;
 import org.apache.fluo.api.exceptions.CommitException;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.core.exceptions.AlreadyAcknowledgedException;
 import org.apache.fluo.core.impl.Environment;
 import org.apache.fluo.core.impl.Notification;
@@ -204,8 +205,8 @@ public class TestTransaction implements TransactionBase {
   }
 
   @Override
-  public ScannerBuilder scanner() {
-    return tx.scanner();
+  public RowIterator get(ScannerConfiguration config) {
+    return tx.get(config);
   }
 
   @Override
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/FluoIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/FluoIT.java
index 086f1d9b..321374b5 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/FluoIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/FluoIT.java
@@ -29,14 +29,15 @@ import org.apache.fluo.api.client.FluoClient;
 import org.apache.fluo.api.client.FluoFactory;
 import org.apache.fluo.api.client.Snapshot;
 import org.apache.fluo.api.client.TransactionBase;
-import org.apache.fluo.api.client.scanner.CellScanner;
 import org.apache.fluo.api.config.FluoConfiguration;
 import org.apache.fluo.api.config.ObserverConfiguration;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumnValue;
 import org.apache.fluo.api.data.Span;
 import org.apache.fluo.api.exceptions.CommitException;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.api.observer.AbstractObserver;
 import org.apache.fluo.core.exceptions.AlreadyAcknowledgedException;
 import org.apache.fluo.core.impl.Environment;
@@ -468,11 +469,14 @@ public class FluoIT extends ITBaseImpl {
     tx3.done();
 
     HashSet<Column> columns = new HashSet<>();
-
-    CellScanner cellScanner =
-        tx2.scanner().over(Span.exact(Bytes.of("d00001"))).fetch(new Column("outlink")).build();
-    for (RowColumnValue rcv : cellScanner) {
-      columns.add(rcv.getColumn());
+    RowIterator riter =
+        tx2.get(new ScannerConfiguration().setSpan(Span.exact(Bytes.of("d00001"),
+            new Column(Bytes.of("outlink")))));
+    while (riter.hasNext()) {
+      ColumnIterator citer = riter.next().getValue();
+      while (citer.hasNext()) {
+        columns.add(citer.next().getKey());
+      }
     }
 
     tx2.done();
@@ -486,12 +490,15 @@ public class FluoIT extends ITBaseImpl {
 
     TestTransaction tx4 = new TestTransaction(env);
     columns.clear();
-    cellScanner =
-        tx4.scanner().over(Span.exact(Bytes.of("d00001"))).fetch(new Column("outlink")).build();
-    for (RowColumnValue rcv : cellScanner) {
-      columns.add(rcv.getColumn());
+    riter =
+        tx4.get(new ScannerConfiguration().setSpan(Span.exact(Bytes.of("d00001"),
+            new Column(Bytes.of("outlink")))));
+    while (riter.hasNext()) {
+      ColumnIterator citer = riter.next().getValue();
+      while (citer.hasNext()) {
+        columns.add(citer.next().getKey());
+      }
     }
-
     expected.add(new Column("outlink", "http://z.com"));
     expected.remove(new Column("outlink", "http://b.com"));
     Assert.assertEquals(expected, columns);
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ScannerIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/ScannerIT.java
deleted file mode 100644
index bdf473d7..00000000
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/ScannerIT.java
+++ /dev/null
@@ -1,217 +0,0 @@
-/*
- * 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.fluo.integration.impl;
-
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Set;
-
-import com.google.common.base.Predicate;
-import com.google.common.base.Predicates;
-import com.google.common.collect.Iterables;
-import org.apache.fluo.api.client.Snapshot;
-import org.apache.fluo.api.client.Transaction;
-import org.apache.fluo.api.client.scanner.CellScanner;
-import org.apache.fluo.api.client.scanner.ColumnScanner;
-import org.apache.fluo.api.client.scanner.RowScanner;
-import org.apache.fluo.api.data.Bytes;
-import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.ColumnValue;
-import org.apache.fluo.api.data.RowColumnValue;
-import org.apache.fluo.api.data.Span;
-import org.apache.fluo.integration.ITBaseImpl;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class ScannerIT extends ITBaseImpl {
-
-  static class ColumnPredicate implements Predicate<RowColumnValue> {
-    Column c;
-
-    ColumnPredicate(Column c) {
-      this.c = c;
-    }
-
-    @Override
-    public boolean apply(RowColumnValue input) {
-      return input.getColumn().equals(c);
-    }
-  }
-
-  static class FamilyPredicate implements Predicate<RowColumnValue> {
-    Bytes fam;
-
-    FamilyPredicate(String family) {
-      this.fam = Bytes.of(family);
-    }
-
-    @Override
-    public boolean apply(RowColumnValue input) {
-      return input.getColumn().getFamily().equals(fam);
-    }
-  }
-
-  static class RowPredicate implements Predicate<RowColumnValue> {
-    Bytes row;
-
-    RowPredicate(String row) {
-      this.row = Bytes.of(row);
-    }
-
-    @Override
-    public boolean apply(RowColumnValue input) {
-      return input.getRow().equals(row);
-    }
-  }
-
-  @Test
-  public void testFiltering() {
-    Set<RowColumnValue> expected = genData();
-
-    HashSet<RowColumnValue> expectedR2 = new HashSet<>();
-    Iterables.addAll(expectedR2, Iterables.filter(expected, new RowPredicate("r2")));
-    Assert.assertEquals(2, expectedR2.size());
-
-
-    HashSet<RowColumnValue> expectedR2c = new HashSet<>();
-    Iterables.addAll(
-        expectedR2c,
-        Iterables.filter(expected,
-            Predicates.and(new RowPredicate("r2"), new ColumnPredicate(new Column("f1", "q2")))));
-    Assert.assertEquals(1, expectedR2c.size());
-
-    HashSet<RowColumnValue> expectedC = new HashSet<>();
-    Iterables.addAll(expectedC,
-        Iterables.filter(expected, new ColumnPredicate(new Column("f1", "q1"))));
-    Assert.assertEquals(2, expectedC.size());
-
-    HashSet<RowColumnValue> expectedCF = new HashSet<>();
-    Iterables.addAll(expectedCF, Iterables.filter(expected, new FamilyPredicate("f2")));
-    Assert.assertEquals(2, expectedCF.size());
-
-    HashSet<RowColumnValue> expectedCols = new HashSet<>();
-    Iterables.addAll(expectedCols, Iterables.filter(expected, Predicates.or(new ColumnPredicate(
-        new Column("f2", "q5")), new ColumnPredicate(new Column("f1", "q1")))));
-    Assert.assertEquals(3, expectedCols.size());
-
-    try (Snapshot snap = client.newSnapshot()) {
-      HashSet<RowColumnValue> actual = new HashSet<>();
-      Iterables.addAll(actual, snap.scanner().over(Span.exact("r2")).build());
-      Assert.assertEquals(expectedR2, actual);
-
-      actual.clear();
-      Iterables.addAll(actual, snap.scanner().over(Span.exact("r2")).fetch(new Column("f1", "q2"))
-          .build());
-      Assert.assertEquals(expectedR2c, actual);
-
-      actual.clear();
-      Iterables.addAll(actual, snap.scanner().fetch(new Column("f1", "q1")).build());
-      Assert.assertEquals(expectedC, actual);
-
-      actual.clear();
-      Iterables.addAll(actual, snap.scanner().fetch(new Column("f2")).build());
-      Assert.assertEquals(expectedCF, actual);
-
-      actual.clear();
-      Iterables.addAll(actual, snap.scanner().fetch(new Column("f2", "q5"), new Column("f1", "q1"))
-          .build());
-      Assert.assertEquals(expectedCols, actual);
-    }
-
-  }
-
-  @Test
-  public void testMultipleIteratorsFromSameRowScanner() {
-    Set<RowColumnValue> expected = genData();
-
-    try (Snapshot snap = client.newSnapshot()) {
-      RowScanner rowScanner = snap.scanner().byRow().build();
-
-      Iterator<ColumnScanner> iter1 = rowScanner.iterator();
-      Iterator<ColumnScanner> iter2 = rowScanner.iterator();
-
-      HashSet<RowColumnValue> actual1 = new HashSet<>();
-      HashSet<RowColumnValue> actual2 = new HashSet<>();
-
-      while (iter1.hasNext()) {
-        ColumnScanner cs1 = iter1.next();
-
-        Assert.assertTrue(iter2.hasNext());
-        ColumnScanner cs2 = iter2.next();
-
-        for (ColumnValue cv : cs1) {
-          actual1.add(new RowColumnValue(cs1.getRow(), cv.getColumn(), cv.getValue()));
-        }
-
-        for (ColumnValue cv : cs2) {
-          actual2.add(new RowColumnValue(cs2.getRow(), cv.getColumn(), cv.getValue()));
-        }
-      }
-
-      Assert.assertFalse(iter2.hasNext());
-
-      Assert.assertEquals(expected, actual1);
-      Assert.assertEquals(expected, actual2);
-    }
-  }
-
-  @Test
-  public void testMultipleIteratorsFromSameIterable() {
-
-    Set<RowColumnValue> expected = genData();
-
-    try (Snapshot snap = client.newSnapshot()) {
-      CellScanner cellScanner = snap.scanner().build();
-      // grab two iterators from same iterable and iterator over them in interleaved fashion
-      Iterator<RowColumnValue> iter1 = cellScanner.iterator();
-      Iterator<RowColumnValue> iter2 = cellScanner.iterator();
-
-      HashSet<RowColumnValue> actual1 = new HashSet<>();
-      HashSet<RowColumnValue> actual2 = new HashSet<>();
-
-      while (iter1.hasNext()) {
-        Assert.assertTrue(iter2.hasNext());
-        actual1.add(iter1.next());
-        actual2.add(iter2.next());
-      }
-
-      Assert.assertFalse(iter2.hasNext());
-
-      Assert.assertEquals(expected, actual1);
-      Assert.assertEquals(expected, actual2);
-    }
-  }
-
-  private Set<RowColumnValue> genData() {
-    Set<RowColumnValue> expected = new HashSet<>();
-    expected.add(new RowColumnValue("r1", new Column("f1", "q1"), "v1"));
-    expected.add(new RowColumnValue("r1", new Column("f2", "q3"), "v2"));
-    expected.add(new RowColumnValue("r2", new Column("f1", "q1"), "v3"));
-    expected.add(new RowColumnValue("r2", new Column("f1", "q2"), "v4"));
-    expected.add(new RowColumnValue("r4", new Column("f2", "q5"), "v5"));
-
-    Assert.assertEquals(5, expected.size());
-
-    try (Transaction tx = client.newTransaction()) {
-      for (RowColumnValue rcv : expected) {
-        tx.set(rcv.getRow(), rcv.getColumn(), rcv.getValue());
-      }
-      tx.commit();
-    }
-
-    return expected;
-  }
-}
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/StochasticBankIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/StochasticBankIT.java
index caa7357e..720bfd8c 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/StochasticBankIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/StochasticBankIT.java
@@ -34,9 +34,12 @@ import org.apache.accumulo.core.client.Scanner;
 import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.util.Stat;
 import org.apache.fluo.accumulo.format.FluoFormatter;
+import org.apache.fluo.api.config.ScannerConfiguration;
+import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumnValue;
 import org.apache.fluo.api.exceptions.CommitException;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.core.impl.Environment;
 import org.apache.fluo.integration.ITBaseImpl;
 import org.apache.fluo.integration.TestTransaction;
@@ -188,11 +191,18 @@ public class StochasticBankIT extends ITBaseImpl {
         long t1 = System.currentTimeMillis();
 
         TestTransaction tx = new TestTransaction(env);
+        RowIterator iter = tx.get(new ScannerConfiguration());
+
         Stat stat = new Stat();
 
-        for (RowColumnValue rcv : tx.scanner().build()) {
-          int amt = Integer.parseInt(rcv.getValue().toString());
+        while (iter.hasNext()) {
+          Entry<Bytes, ColumnIterator> colIter = iter.next();
+          Entry<Column, Bytes> column = colIter.getValue().next();
+
+          int amt = Integer.parseInt(column.getValue().toString());
+
           stat.addStat(amt);
+
         }
 
         long t2 = System.currentTimeMillis();
@@ -263,8 +273,12 @@ public class StochasticBankIT extends ITBaseImpl {
   private static HashMap<String, String> toMap(TestTransaction tx) throws Exception {
     HashMap<String, String> map = new HashMap<>();
 
-    for (RowColumnValue rcv : tx.scanner().build()) {
-      map.put(rcv.getRow().toString(), rcv.getValue().toString());
+    RowIterator iter = tx.get(new ScannerConfiguration());
+    while (iter.hasNext()) {
+      Entry<Bytes, ColumnIterator> colIter = iter.next();
+      Entry<Column, Bytes> column = colIter.getValue().next();
+
+      map.put(colIter.getKey().toString(), column.getValue().toString());
     }
 
     return map;
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/WeakNotificationIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/WeakNotificationIT.java
index de9ab41a..ea5921c2 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/WeakNotificationIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/WeakNotificationIT.java
@@ -17,15 +17,17 @@ package org.apache.fluo.integration.impl;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Map.Entry;
 
 import org.apache.fluo.api.client.Transaction;
 import org.apache.fluo.api.client.TransactionBase;
-import org.apache.fluo.api.client.scanner.CellScanner;
 import org.apache.fluo.api.config.ObserverConfiguration;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
-import org.apache.fluo.api.data.RowColumnValue;
 import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.api.observer.AbstractObserver;
 import org.apache.fluo.core.impl.Environment;
 import org.apache.fluo.core.impl.TransactionImpl.CommitData;
@@ -46,14 +48,19 @@ public class WeakNotificationIT extends ITBaseMini {
     @Override
     public void process(TransactionBase tx, Bytes row, Column col) throws Exception {
 
-      CellScanner cellScanner =
-          tx.scanner().over(Span.exact(row, new Column(Bytes.of("stats")))).build();
+      ScannerConfiguration sc = new ScannerConfiguration();
+      sc.setSpan(Span.exact(row, new Column(Bytes.of("stats"))));
+      RowIterator rowIter = tx.get(sc);
 
       int sum = 0;
 
-      for (RowColumnValue rcv : cellScanner) {
-        sum += Integer.parseInt(rcv.getValue().toString());
-        tx.delete(row, rcv.getColumn());
+      if (rowIter.hasNext()) {
+        ColumnIterator colIter = rowIter.next().getValue();
+        while (colIter.hasNext()) {
+          Entry<Column, Bytes> colVal = colIter.next();
+          sum += Integer.parseInt(colVal.getValue().toString());
+          tx.delete(row, colVal.getKey());
+        }
       }
 
       if (sum != 0) {
diff --git a/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java b/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
index d1d4522e..e4204aae 100644
--- a/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
+++ b/modules/integration/src/test/java/org/apache/fluo/integration/impl/WorkerIT.java
@@ -18,15 +18,17 @@ package org.apache.fluo.integration.impl;
 import java.util.Collections;
 import java.util.List;
 
-import com.google.common.collect.Iterables;
 import org.apache.fluo.api.client.Snapshot;
 import org.apache.fluo.api.client.Transaction;
 import org.apache.fluo.api.client.TransactionBase;
 import org.apache.fluo.api.config.ObserverConfiguration;
+import org.apache.fluo.api.config.ScannerConfiguration;
 import org.apache.fluo.api.data.Bytes;
 import org.apache.fluo.api.data.Column;
 import org.apache.fluo.api.data.RowColumn;
 import org.apache.fluo.api.data.Span;
+import org.apache.fluo.api.iterator.ColumnIterator;
+import org.apache.fluo.api.iterator.RowIterator;
 import org.apache.fluo.api.observer.Observer;
 import org.apache.fluo.core.impl.Environment;
 import org.apache.fluo.core.impl.TransactionImpl.CommitData;
@@ -69,8 +71,17 @@ public class WorkerIT extends ITBaseMini {
       String degree = tx.gets(row, DEGREE);
 
       // calculate new degree
-      String degree2 =
-          "" + Iterables.size(tx.scanner().over(Span.exact(row, new Column("link"))).build());
+      int count = 0;
+      RowIterator riter =
+          tx.get(new ScannerConfiguration().setSpan(Span.exact(row, new Column("link"))));
+      while (riter.hasNext()) {
+        ColumnIterator citer = riter.next().getValue();
+        while (citer.hasNext()) {
+          citer.next();
+          count++;
+        }
+      }
+      String degree2 = "" + count;
 
       if (degree == null || !degree.equals(degree2)) {
         tx.set(row, DEGREE, degree2);