You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@accumulo.apache.org by echeipesh <gi...@git.apache.org> on 2015/04/01 00:56:52 UTC
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
GitHub user echeipesh opened a pull request:
https://github.com/apache/accumulo/pull/25
ACCUMULO-3602 BatchScanner optimization for AccumuloInputFormat
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/echeipesh/accumulo ACCUMULO-3602
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/accumulo/pull/25.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #25
----
commit 8da593302e2c702d8a775752d5f5620d2ddb3345
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-23T20:43:46Z
Added support for grouping ranges per tablet when using AccumuloInputFormat
commit 47a2fae8423ab383364228469055904803751c1a
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-25T15:08:20Z
Avoid casts by using AccumuloInputSplit interface
commit b4101a37949271a1b388ac5bd3f1b1d14e00d272
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-25T15:10:05Z
upgrade warning to exception if batch scan is requested but not available
commit ad83769d96224c55f909ae4ba0d6482c7555bf1c
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-25T15:11:24Z
close underlying scanner with the RecordReader
commit 46a45c3582289730905ee4a68fa3fcf8f5245ccd
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T19:08:37Z
Upgrade AccumuloInputSplit to abstract class for code reuse
commit 7a09974b4ae731c78e7a99ec94d02c3a7a5ad0b1
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T19:53:58Z
merging upstream master
commit 41e79cf9b71b48be9acf2a44a44a523476c45024
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T20:00:42Z
fix merge errors
commit 7b3beebb2ba023e8862f45c89b61435a7b404368
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T20:32:06Z
add batch scan setters/getters to AccumuloInputFormat
commit c84c8c035414d6f32a864b573f7f6a1f6c16551a
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T21:47:17Z
fix casting error
commit 35495142d007303cbba3c083a07abbe1a859e289
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T21:47:35Z
make findbugs happy
commit 7ac1cef68d651b9c19bfc57db28a718630ae37d1
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T21:47:48Z
test BatchInputSplit generation
commit 9a477220a205816aa8641d2b362c3b2c395c83f0
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T21:58:26Z
match code flow between mapred and mapreduce
also: killing dead code for “iterators”
commit 23f6d400b4794ea8c30570f51fcdae09ba33a17c
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T22:21:59Z
test all splits for correct type, not just head
commit 17fa276c8dc06c6036bc7a4c9ccc2a8cf23f90e9
Author: Eugene Cheipesh <ec...@gmail.com>
Date: 2015-03-31T22:55:32Z
Test MR job with BatchScan option enable
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28597474
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
Actually Auto-adjust is on by default, which would cause an exception as soon as batch scan is turned on. Should it be downgraded to a warning, since interaction between auto-adjust and batch scan is a "soft" change in the behavior?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27924910
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java ---
@@ -517,6 +517,74 @@ public static Boolean isOfflineScan(Class<?> implementingClass, Configuration co
}
/**
+ * Controls the use of the {@link BatchScanner} in this job.
+ * Using this feature will group ranges by their source tablet per InputSplit and use BatchScanner to read them.
+ *
+ * <p>
+ * By default, this feature is <b>disabled</b>.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @param enableFeature
+ * the feature is enabled if true, disabled otherwise
+ * @since 1.7.0
+ */
+ public static void setBatchScan(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
+ conf.setBoolean(enumToConfKey(implementingClass, Features.BATCH_SCANNER), enableFeature);
+ }
+
+ /**
+ * Determines whether a configuration has the BatchScanner feature enabled.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @return true if the feature is enabled, false otherwise
+ * @since 1.7.0
+ * @see #setBatchScan(Class, Configuration, boolean)
+ */
+ public static Boolean isBatchScan(Class<?> implementingClass, Configuration conf) {
+ return conf.getBoolean(enumToConfKey(implementingClass, Features.BATCH_SCANNER), false);
+ }
+
+ /**
+ * Controls the use of the {@link BatchScanner} in this job.
+ * Using this feature will group ranges by their source tablet per InputSplit and use BatchScanner to read them.
+ *
+ * <p>
+ * By default, this feature is <b>disabled</b>.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @param count
+ * number of threads to use with BatchScanner
+ * @since 1.7.0
+ */
+ public static void setBatchScanThreads(Class<?> implementingClass, Configuration conf, int count) {
--- End diff --
> Should it still remain for future refactors where BatchInputSplit gets initialized with ranges that span tablets?
Could remove the methods for now and a future changes could add them if needed. Having them the way things work now is misleading.
> Perhaps reduce default to one thread for now?
If the methods were removed, the implementation could do that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-94060780
Actually, compat is still broken with that setupIterators method. Do you want to fix that @echeipesh or should I just do when I apply the rest of your changes?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27628662
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -625,32 +659,52 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan && supportBatchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
+ clippedRanges.add(ke.clip(r));
+
+ org.apache.accumulo.core.client.mapred.BatchInputSplit split = new org.apache.accumulo.core.client.mapred.BatchInputSplit(tableName, tableId, clippedRanges, new String[] {location});
+ split.setFetchedColumns(tableConfig.getFetchedColumns());
--- End diff --
Can we set the common attributes on the splits in one place instead of both sides of the conditional? Does the type hierarchy make that simple?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27628925
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputSplit.java ---
@@ -0,0 +1,425 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.core.client.mapreduce;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.ClientConfiguration;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.mapreduce.lib.impl.ConfiguratorBase.TokenSource;
+import org.apache.accumulo.core.client.mapreduce.lib.impl.InputConfigurator;
+import org.apache.accumulo.core.client.mock.MockInstance;
+import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
+import org.apache.accumulo.core.client.security.tokens.AuthenticationToken.AuthenticationTokenSerializer;
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.Base64;
+import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.core.data.Key;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.log4j.Level;
+
+/**
+ * Abstracts over configurations common to all InputSplits. Specifically it leaves out methods
+ * related to number of ranges and locations per InputSplit as those vary by implementation.
+ *
+ * @see RangeInputSplit
+ * @see BatchInputSplit
+ */
+public abstract class AccumuloInputSplit extends InputSplit implements Writable {
--- End diff --
Like BatchInputSplit, I think we'd want to put this in an impl/ package.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28555436
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java ---
@@ -315,19 +352,8 @@ protected static TabletLocator getTabletLocator(JobConf job) throws TableNotFoun
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
@Override
- protected void setupIterators(JobConf job, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapred.RangeInputSplit split) {
- List<IteratorSetting> iterators = null;
-
- if (null == split) {
- iterators = getIterators(job);
- } else {
- iterators = split.getIterators();
- if (null == iterators) {
- iterators = getIterators(job);
- }
- }
-
- setupIterators(iterators, scanner);
+ protected List<IteratorSetting> jobIterators(JobConf job, String tableName) {
+ return getIterators(job);
--- End diff --
n/m, pulled it into Eclipse and realized it's just me :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28607317
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
Sorry for confusing you guys, it's actually backwards. BatchScan does essentially the same thing as AutoAdjust in terms of clipping on tablet boundaries. AutoAdjust is on by default, so there is no issue. Asking for BatchScan and AutoAdjust off is a nonsensical case.
So I've added mentions in both javadocs as Keith suggested and an exception on AutoAdjust = false BatchScan = true, since it indicates user specifically asked for something we can not and know not how to give them.
This is pushed up now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27974261
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -313,11 +374,34 @@ protected static TabletLocator getTabletLocator(JobContext context) throws Table
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
+ /**
+ * Apply the configured iterators from the configuration to the scanner for the specified table name
+ *
+ * @param context
+ * the Hadoop context for the configured job
+ * @param scanner
+ * the scanner to configure
+ * @since 1.7.0
+ */
@Override
- protected void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapreduce.RangeInputSplit split) {
+ protected void setupIterators(TaskAttemptContext context, ScannerBase scanner, String tableName, AccumuloInputSplit split) {
--- End diff --
another usage of AccumuloInputSplit
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27926398
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java ---
@@ -517,6 +517,74 @@ public static Boolean isOfflineScan(Class<?> implementingClass, Configuration co
}
/**
+ * Controls the use of the {@link BatchScanner} in this job.
+ * Using this feature will group ranges by their source tablet per InputSplit and use BatchScanner to read them.
--- End diff --
Could mention what options this will not work with in javadoc. For example setting this will not work w/ offline scan and local iters.
Could also mention the benefit of doing this. It seems the main benefit is one RPC to send multiple ranges to a tserver and then read the results, versus an RPC per range.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27986006
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -384,7 +387,21 @@ public static InputTableConfig getInputTableConfig(JobConf job, String tableName
protected abstract static class AbstractRecordReader<K,V> implements RecordReader<K,V> {
protected long numKeysRead;
protected Iterator<Map.Entry<Key,Value>> scannerIterator;
- protected RangeInputSplit split;
+ protected org.apache.accumulo.core.client.mapreduce.impl.AccumuloInputSplit split;
+ protected ScannerBase scannerBase;
+
+ /**
+ * Configures the iterators on a scanner for the given table name.
+ *
+ * @param job
+ * the Hadoop job configuration
+ * @param scanner
+ * the scanner for which to configure the iterators
+ * @param tableName
+ * the table name for which the scanner is configured
+ * @since 1.7.0
+ */
+ protected abstract void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split);
--- End diff --
I like how your suggestion moves configuration of the scanner to a private method. Trying to understand your suggestion made me realize I don't understand why `AccumuloInputSplit` is storing and serializing things like iterators. Why not get that info from the job conf when the task runs? I am being lazy asking this question, let me know if its not quick/easy to explain I can dig around in the code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-88596902
No worries, I was just looking through this, I see what happened.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27914201
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan && supportBatchScan) {
--- End diff --
Why check `supportBatchScan` here?. There is a check earlier that throws an exception `if (batchScan && !supportBatchScan)`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27927193
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -384,7 +387,21 @@ public static InputTableConfig getInputTableConfig(JobConf job, String tableName
protected abstract static class AbstractRecordReader<K,V> implements RecordReader<K,V> {
protected long numKeysRead;
protected Iterator<Map.Entry<Key,Value>> scannerIterator;
- protected RangeInputSplit split;
+ protected org.apache.accumulo.core.client.mapreduce.impl.AccumuloInputSplit split;
+ protected ScannerBase scannerBase;
+
+ /**
+ * Configures the iterators on a scanner for the given table name.
+ *
+ * @param job
+ * the Hadoop job configuration
+ * @param scanner
+ * the scanner for which to configure the iterators
+ * @param tableName
+ * the table name for which the scanner is configured
+ * @since 1.7.0
+ */
+ protected abstract void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split);
--- End diff --
I suppose `setupIterators` would be considered public API. However it has parameter of type `AccumuloInputSplit` which is not public.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27988729
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -384,7 +387,21 @@ public static InputTableConfig getInputTableConfig(JobConf job, String tableName
protected abstract static class AbstractRecordReader<K,V> implements RecordReader<K,V> {
protected long numKeysRead;
protected Iterator<Map.Entry<Key,Value>> scannerIterator;
- protected RangeInputSplit split;
+ protected org.apache.accumulo.core.client.mapreduce.impl.AccumuloInputSplit split;
+ protected ScannerBase scannerBase;
+
+ /**
+ * Configures the iterators on a scanner for the given table name.
+ *
+ * @param job
+ * the Hadoop job configuration
+ * @param scanner
+ * the scanner for which to configure the iterators
+ * @param tableName
+ * the table name for which the scanner is configured
+ * @since 1.7.0
+ */
+ protected abstract void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split);
--- End diff --
Actually that's just factored out of old `RangeInputSplit`. Looking over the code there doesn't seem to be a reason for it. All of those options are coming from `InputTableConfig` which, as you say, is deserialized from `JobContext`. Looking over the code there really isn't any hard reason for that (and most fields in `AccumuloInputSplit`) to be there. Although I think it makes it a little easier to reason about.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-94108040
@joshelser, awesome! Thank you for all the help guiding this along.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28567534
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
If Eugene agrees with me, yes I believe so.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28168170
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloMultiTableInputFormat.java ---
@@ -86,15 +87,8 @@ public boolean nextKeyValue() throws IOException, InterruptedException {
}
@Override
- protected void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName, RangeInputSplit split) {
- List<IteratorSetting> iterators = split.getIterators();
- if (null == iterators) {
- iterators = getInputTableConfig(context, tableName).getIterators();
- }
-
- for (IteratorSetting setting : iterators) {
- scanner.addScanIterator(setting);
- }
+ protected List<IteratorSetting> contextIterators(TaskAttemptContext context, String tableName) {
+ return getInputTableConfig(context, tableName).getIterators();
--- End diff --
I notice that `mapred.MultiTableInputFormat` does not have similar logic but just uses Reader from `InputFormatBase` which appears to assume there is only one table target:
https://github.com/apache/accumulo/blob/master/core/src/main/java/org/apache/accumulo/core/client/mapred/AccumuloMultiTableInputFormat.java#L68
I am not sure if this is a bug, as it looks like, or there is some reason for this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-90646891
Looks like it's called by the framework by wrapping the `RecordReader` from the `InputSplit`, wrapping it in another `RecordReader` and then calling `getProgress()` during `nextKeyValue()`. I think this means that it's OK.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-93758183
Summary of changes since last burst of comments:
- Removed `AccumuloInputSplit` references in public API via method presented
- Removed `.setBatchScanThreads` from configuration
- Added `.setBatchScan` option to `mapred` InputFormat
- Added better scaladoc for `.setBatchScan`
- Covered all changes with `@depricated` since and `@since` tags
- Made small improvement to a test to verify resumption of success after testing failure.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27628961
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/BatchInputSplit.java ---
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.core.client.mapreduce;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+
+/**
+ * The Class BatchInputSplit. Encapsulates a set of Accumulo ranges on a single tablet for use in Map Reduce jobs.
+ * Can contain several Ranges per split.
+ */
+public class BatchInputSplit extends AccumuloInputSplit {
--- End diff --
Ditto impl package
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27629947
--- Diff: test/src/test/java/org/apache/accumulo/test/functional/AccumuloInputFormatIT.java ---
@@ -146,6 +147,23 @@ public void testGetSplits() throws IOException, AccumuloSecurityException, Accum
AccumuloInputFormat.setAutoAdjustRanges(job, false);
splits = inputFormat.getSplits(job);
assertEquals(ranges.size(), splits.size());
+
+ //BatchScan not available for offline scans
+ AccumuloInputFormat.setBatchScan(job, true);
--- End diff --
Maybe expand this same test for isolated scans local iterators too?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27974024
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java ---
@@ -315,7 +317,11 @@ protected static TabletLocator getTabletLocator(JobConf job) throws TableNotFoun
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
@Override
- protected void setupIterators(JobConf job, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapred.RangeInputSplit split) {
+ protected void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split) {
--- End diff --
There's a since tag on AbstractRecordReader. Isn't that sufficient?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27974211
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -313,11 +374,34 @@ protected static TabletLocator getTabletLocator(JobContext context) throws Table
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
+ /**
+ * Apply the configured iterators from the configuration to the scanner for the specified table name
+ *
+ * @param context
+ * the Hadoop context for the configured job
+ * @param scanner
+ * the scanner to configure
+ * @since 1.7.0
+ */
@Override
- protected void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapreduce.RangeInputSplit split) {
--- End diff --
should this be deprecated instead of deleted?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27973581
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java ---
@@ -315,7 +317,11 @@ protected static TabletLocator getTabletLocator(JobConf job) throws TableNotFoun
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
@Override
- protected void setupIterators(JobConf job, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapred.RangeInputSplit split) {
+ protected void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split) {
--- End diff --
* should have a since tag
* AccumuloInputSplit is not public
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28556681
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -325,27 +362,12 @@ protected void setupIterators(TaskAttemptContext context, Scanner scanner, Strin
* the Hadoop context for the configured job
* @param scanner
* the scanner to configure
+ * @deprecated since 1.7.0; Use {@link #contextIterators} instead.
*/
@Deprecated
protected void setupIterators(TaskAttemptContext context, Scanner scanner) {
- setupIterators(context, scanner, null);
- }
-
- /**
- * Initialize a scanner over the given input split using this task attempt configuration.
- */
- protected void setupIterators(TaskAttemptContext context, Scanner scanner, org.apache.accumulo.core.client.mapreduce.RangeInputSplit split) {
--- End diff --
This breaks compatibility with 1.6. Needs to be re-introduced.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27629761
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/RangeInputSplit.java ---
@@ -179,81 +93,13 @@ public void readFields(DataInput in) throws IOException {
if (in.readBoolean()) {
localIterators = in.readBoolean();
}
-
- if (in.readBoolean()) {
- mockInstance = in.readBoolean();
- }
-
- if (in.readBoolean()) {
- int numColumns = in.readInt();
- List<String> columns = new ArrayList<String>(numColumns);
- for (int i = 0; i < numColumns; i++) {
- columns.add(in.readUTF());
- }
-
- fetchedColumns = InputConfigurator.deserializeFetchedColumns(columns);
- }
-
- if (in.readBoolean()) {
- String strAuths = in.readUTF();
- auths = new Authorizations(strAuths.getBytes(UTF_8));
- }
-
- if (in.readBoolean()) {
- principal = in.readUTF();
- }
-
- if (in.readBoolean()) {
- int ordinal = in.readInt();
- this.tokenSource = TokenSource.values()[ordinal];
-
- switch (this.tokenSource) {
- case INLINE:
- String tokenClass = in.readUTF();
- byte[] base64TokenBytes = in.readUTF().getBytes(UTF_8);
- byte[] tokenBytes = Base64.decodeBase64(base64TokenBytes);
-
- this.token = AuthenticationTokenSerializer.deserialize(tokenClass, tokenBytes);
- break;
-
- case FILE:
- this.tokenFile = in.readUTF();
-
- break;
- default:
- throw new IOException("Cannot parse unknown TokenSource ordinal");
- }
- }
-
- if (in.readBoolean()) {
- instanceName = in.readUTF();
- }
-
- if (in.readBoolean()) {
- zooKeepers = in.readUTF();
- }
-
- if (in.readBoolean()) {
- int numIterators = in.readInt();
- iterators = new ArrayList<IteratorSetting>(numIterators);
- for (int i = 0; i < numIterators; i++) {
- iterators.add(new IteratorSetting(in));
- }
- }
-
- if (in.readBoolean()) {
- level = Level.toLevel(in.readInt());
- }
}
@Override
public void write(DataOutput out) throws IOException {
--- End diff --
Need to serialize offline here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27978633
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -397,15 +414,17 @@ public static InputTableConfig getInputTableConfig(JobConf job, String tableName
* the table name for which the scanner is configured
* @since 1.6.0
*/
- protected abstract void setupIterators(JobConf job, Scanner scanner, String tableName, RangeInputSplit split);
+ @Deprecated
--- End diff --
Another common practice in the javadoc is to provide a pointer to the replacement for the deprecated method if one exists.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27983742
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -384,7 +387,21 @@ public static InputTableConfig getInputTableConfig(JobConf job, String tableName
protected abstract static class AbstractRecordReader<K,V> implements RecordReader<K,V> {
protected long numKeysRead;
protected Iterator<Map.Entry<Key,Value>> scannerIterator;
- protected RangeInputSplit split;
+ protected org.apache.accumulo.core.client.mapreduce.impl.AccumuloInputSplit split;
+ protected ScannerBase scannerBase;
+
+ /**
+ * Configures the iterators on a scanner for the given table name.
+ *
+ * @param job
+ * the Hadoop job configuration
+ * @param scanner
+ * the scanner for which to configure the iterators
+ * @param tableName
+ * the table name for which the scanner is configured
+ * @since 1.7.0
+ */
+ protected abstract void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split);
--- End diff --
That's a good point. In an effort to avoid mistakes of the past and keep AccumulInputSplit out of public API how about something like this:
```java
// InputFormatBase
List<IteratorSetting> contextIterators(TaskAttemptContext context, String tableName) {
return getIterators(context).getIterators();
}
// AccumululoMultiTableInputFormat
List<IteratorSetting> contextIterators(TaskAttemptContext context, String tableName) {
return getInputTableConfig(context, tableName).getIterators();
}
// lift setupIterators to AbstractInputFormat (no longer abstract)
protected abstract List<IteratorSetting> contextIterators(TaskAttemptContext context, String tableName);
private void setupIterators(TaskAttemptContext context, ScannerBase scanner, String tableName, AccumuloInputSplit split) {
List<IteratorSetting> iterators = null;
if (null == split) {
iterators = contextIterators(context);
} else {
iterators = split.getIterators();
if (null == iterators) {
iterators = contextIterators(context);
}
}
for (IteratorSetting iterator : iterators)
scanner.addScanIterator(iterator);
}
```
Looking at code right now the only part that diverges is pulling out default split if they're somehow unavailable in the split. This narrows the interface, hopefully in an acceptable way?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-94032333
Hot diggity. I'll see if I can bring this down, run some of the UTs/ITs, squash the commits, and push it up. Thanks so much for your hard work, @echeipesh!!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91033820
I was discussing the big picture behind this PR w/ @ctubbsii . It seems like this change could encourage users to pass many ranges as configuration for the map reduce job. This could cause memory exhaustion for the job tracker.
We discussed passing a function which generates a set of ranges, instead of passing lots of ranges. The implementation would still use a batch scanner (or scanner with a special iterator but its harder to pass code to tserver). Each input split could call a function like the following which deterministically creates a set of ranges. Then those ranges could be used for the batch scanner.
```java
interface RangeGenerator {
/**
* @param tabletRange The data range for the tablet over which the input split is executing
* @param config a mysterious class that allows user to pass parameters to the function
*/
List<Range> createRanges(Range tabletRange, Myst config);
}
```
When configuring the AccumuloInputFormat to use the batch scanner, a class name that implements this function would be provided. The ranges set on the job would be large ranges for portions of table to process. An input split would be created for each tablet that falls within those large ranges, and for each input split the function would be called to possibly create many more ranges.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28560936
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
Should anything need to be added the `setBatchScan(...)` javadoc regarding `autoAdjust`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh closed the pull request at:
https://github.com/apache/accumulo/pull/25
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-90621844
I believe it has to do with catching the generic Exception. https://github.com/echeipesh/accumulo/pull/1/files made it pass for me locally.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91053066
> then only we only implement it and have less code to maintain.
Gentle reminder that you're looking a gift horse in the mouth with this. Eugene has done a lot of work here as a first contribution. If there's nothing technically wrong with this approach, he should be rewarded for that. If there is a simplification that you think should be made, you are able to do so yourself. I feel like this is not being sufficiently represented in this discussion.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27629663
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -323,7 +382,7 @@ protected static TabletLocator getTabletLocator(JobContext context) throws Table
* @since 1.6.0
*/
@Override
- protected void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapreduce.RangeInputSplit split) {
+ protected void setupIterators(TaskAttemptContext context, ScannerBase scanner, String tableName, org.apache.accumulo.core.client.mapreduce.AccumuloInputSplit split) {
--- End diff --
Sorry, we can't change these as they'd break API from 1.6.0. We can add to the API for 1.7.0, but we can't remove. Need to deprecate the ones that take Scanner and add in a new one for ScannerBase.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-90642462
:+1: This looks great! I'm going to take a quick glance at the mapreduce code and see if I can make myself more comfortable about the potential concurrency issue I mentioned.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by ctubbsii <gi...@git.apache.org>.
Github user ctubbsii commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91038876
> Agreed, but is this a new issue?
I wouldn't think so. I think it highlights a better implementation to satisfy the underlying use case than the one currently being proposed in this PR.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27914471
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan && supportBatchScan) {
--- End diff --
Mostly for code clarity, that there are two things that must be in line. Although, as you say, functionally checking for batchScan only would be sufficient.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27973707
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java ---
@@ -315,7 +317,11 @@ protected static TabletLocator getTabletLocator(JobConf job) throws TableNotFoun
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
@Override
- protected void setupIterators(JobConf job, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapred.RangeInputSplit split) {
+ protected void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split) {
+ setupIterators(job, scanner, split);
+ }
+
+ protected void setupIterators(JobConf job, ScannerBase scanner, AccumuloInputSplit split) {
--- End diff --
needs since tag
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27981389
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
The intent of the change is to keep splits local and batch access to them to reduce overhead. This implies clipping the ranges. The alternative could be a split with two locations that has 90%+ of ranges on vs the other location. As I understand the scheduler at that point has no weight for locality preference.
On other hand having option to not clip would make the options more logically consistent and justify having number of threads per batch reader as configuration.
I personally think the first option breaks the least eggs, what do you think?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27921845
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java ---
@@ -517,6 +517,74 @@ public static Boolean isOfflineScan(Class<?> implementingClass, Configuration co
}
/**
+ * Controls the use of the {@link BatchScanner} in this job.
+ * Using this feature will group ranges by their source tablet per InputSplit and use BatchScanner to read them.
+ *
+ * <p>
+ * By default, this feature is <b>disabled</b>.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @param enableFeature
+ * the feature is enabled if true, disabled otherwise
+ * @since 1.7.0
+ */
+ public static void setBatchScan(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
+ conf.setBoolean(enumToConfKey(implementingClass, Features.BATCH_SCANNER), enableFeature);
+ }
+
+ /**
+ * Determines whether a configuration has the BatchScanner feature enabled.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @return true if the feature is enabled, false otherwise
+ * @since 1.7.0
+ * @see #setBatchScan(Class, Configuration, boolean)
+ */
+ public static Boolean isBatchScan(Class<?> implementingClass, Configuration conf) {
+ return conf.getBoolean(enumToConfKey(implementingClass, Features.BATCH_SCANNER), false);
+ }
+
+ /**
+ * Controls the use of the {@link BatchScanner} in this job.
+ * Using this feature will group ranges by their source tablet per InputSplit and use BatchScanner to read them.
+ *
+ * <p>
+ * By default, this feature is <b>disabled</b>.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @param count
+ * number of threads to use with BatchScanner
+ * @since 1.7.0
+ */
+ public static void setBatchScanThreads(Class<?> implementingClass, Configuration conf, int count) {
--- End diff --
The batch scanner will not use more than one thread per tablet. If each mapper is using a batch scanner to read multiple ranges from a single tablet, then the batch scanner will only use a single thread to do this even if it has more threads it could use.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28601270
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
Maybe only make a javadoc change regarding this? Just state that auto adjust is ignored in `setBatchScan` and `autoAdjust` javadoc?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-94077963
And, if you can @echeipesh, can you please close this? Because I squashed and didn't include the magic github markdown, I can't close the PR myself.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27923743
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/lib/impl/InputConfigurator.java ---
@@ -517,6 +517,74 @@ public static Boolean isOfflineScan(Class<?> implementingClass, Configuration co
}
/**
+ * Controls the use of the {@link BatchScanner} in this job.
+ * Using this feature will group ranges by their source tablet per InputSplit and use BatchScanner to read them.
+ *
+ * <p>
+ * By default, this feature is <b>disabled</b>.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @param enableFeature
+ * the feature is enabled if true, disabled otherwise
+ * @since 1.7.0
+ */
+ public static void setBatchScan(Class<?> implementingClass, Configuration conf, boolean enableFeature) {
+ conf.setBoolean(enumToConfKey(implementingClass, Features.BATCH_SCANNER), enableFeature);
+ }
+
+ /**
+ * Determines whether a configuration has the BatchScanner feature enabled.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @return true if the feature is enabled, false otherwise
+ * @since 1.7.0
+ * @see #setBatchScan(Class, Configuration, boolean)
+ */
+ public static Boolean isBatchScan(Class<?> implementingClass, Configuration conf) {
+ return conf.getBoolean(enumToConfKey(implementingClass, Features.BATCH_SCANNER), false);
+ }
+
+ /**
+ * Controls the use of the {@link BatchScanner} in this job.
+ * Using this feature will group ranges by their source tablet per InputSplit and use BatchScanner to read them.
+ *
+ * <p>
+ * By default, this feature is <b>disabled</b>.
+ *
+ * @param implementingClass
+ * the class whose name will be used as a prefix for the property configuration key
+ * @param conf
+ * the Hadoop configuration object to configure
+ * @param count
+ * number of threads to use with BatchScanner
+ * @since 1.7.0
+ */
+ public static void setBatchScanThreads(Class<?> implementingClass, Configuration conf, int count) {
--- End diff --
In the current use case it makes the option unnecessary. Should it still remain for future refactors where `BatchInputSplit` gets initialized with ranges that span tablets? In theory each split could span tservers as well, but it's hard to imagine a reason for that. Perhaps reduce default to one thread for now?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27976594
--- Diff: test/src/test/java/org/apache/accumulo/test/functional/AccumuloInputFormatIT.java ---
@@ -146,6 +146,41 @@ public void testGetSplits() throws IOException, AccumuloSecurityException, Accum
AccumuloInputFormat.setAutoAdjustRanges(job, false);
splits = inputFormat.getSplits(job);
assertEquals(ranges.size(), splits.size());
+
+ //BatchScan not available for offline scans
+ AccumuloInputFormat.setBatchScan(job, true);
+
+ AccumuloInputFormat.setOfflineTableScan(job, true);
+ try {
+ inputFormat.getSplits(job);
+ fail("An exception should have been thrown");
+ } catch (IOException e) {}
+ AccumuloInputFormat.setOfflineTableScan(job, false);
+
+ //BatchScan not available with isolated iterators
+ AccumuloInputFormat.setScanIsolation(job, true);
+ try {
+ inputFormat.getSplits(job);
+ fail("An exception should have been thrown");
+ } catch (IOException e) {}
--- End diff --
nevermind, I see `AccumuloInputFormat.setOfflineTableScan(job, false)` now.. I missed that earlier
Could make the test slight stronger by asserting `getSplits()` does not fail after calling `AccumuloInputFormat.setOfflineTableScan(job, false)`. This would detect any subsequent failures caused by inputFormat holding state related to the 1st failure.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28555268
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -296,6 +296,43 @@ protected static boolean isOfflineScan(JobContext context) {
}
/**
+ * Controls the use of the {@link org.apache.accumulo.core.client.BatchScanner} in this job.
--- End diff --
This is *very* nice!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27826429
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java ---
@@ -426,6 +442,7 @@ protected static ClientConfiguration getClientConfiguration(JobContext context)
* the table name for which the scanner is configured
* @since 1.6.0
*/
+ @Deprecated
protected abstract void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName, RangeInputSplit split);
--- End diff --
I think we could have source compatibility with a switch from Scanner to ScannerBase, but I can't think of a way to handle AccumuloInputSplit instead of only RangeInputSplits. It should be straightforward to just call a new `setupIterator` method from the existing (now deprecated) method, no? Are you just trying to reduce the amount of code in here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27896257
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/impl/BatchInputSplit.java ---
@@ -0,0 +1,166 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.core.client.mapreduce.impl;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+
+/**
+ * The Class BatchInputSplit. Encapsulates a set of Accumulo ranges on a single tablet for use in Map Reduce jobs.
+ * Can contain several Ranges per split.
+ */
+public class BatchInputSplit extends AccumuloInputSplit {
+ private Collection<Range> ranges;
+ private int scanThreads;
+ private float[] rangeProgress = null;
+
+ public BatchInputSplit() {
+ ranges = Collections.emptyList();
+ }
+
+ public BatchInputSplit(BatchInputSplit split) throws IOException {
+ super(split);
+ this.setRanges(split.getRanges());
+ }
+
+ public BatchInputSplit(String table, String tableId, Collection<Range> ranges, String[] locations) {
+ super(table, tableId, locations);
+ this.ranges = ranges;
+ }
+
+ /**
+ * Save progress on each call to this function, implied by value of currentKey, and return average ranges in the split
+ */
+ public float getProgress(Key currentKey) {
+ if (null == rangeProgress)
+ rangeProgress = new float[ranges.size()];
--- End diff --
I'm not sure if this can be called concurrently through the RecordReader. Might need to be synchronized? I need to dig into mapred code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28601820
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
That is a good point. I agree: a warning instead of an error is good enough for me.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27826598
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java ---
@@ -426,6 +442,7 @@ protected static ClientConfiguration getClientConfiguration(JobContext context)
* the table name for which the scanner is configured
* @since 1.6.0
*/
+ @Deprecated
protected abstract void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName, RangeInputSplit split);
--- End diff --
Yes, exactly. The only benefit would be to push up the call up the hierarchy and not have to bother with it in InputFormatBase and AccumuloMultiTableInputFormat.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by wjsl <gi...@git.apache.org>.
Github user wjsl commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91049930
> True in some cases. I was thinking in many cases that a user provided function could deterministically generate a large number of ranges from a small amount of configuration.
That's orthogonal to accepting arbitrary ranges, which doesn't seem to be the focus of this addition. There's not really a reason to say, "we can only have an optimization for range locality IF we restrict the number of ranges a user can input."
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-94018459
:+1: from me
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27979336
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java ---
@@ -315,7 +317,11 @@ protected static TabletLocator getTabletLocator(JobConf job) throws TableNotFoun
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
@Override
- protected void setupIterators(JobConf job, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapred.RangeInputSplit split) {
+ protected void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split) {
--- End diff --
> There's a since tag on AbstractRecordReader. Isn't that sufficient?
oic, yeah thats good.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27973658
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java ---
@@ -315,7 +317,11 @@ protected static TabletLocator getTabletLocator(JobConf job) throws TableNotFoun
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
@Override
- protected void setupIterators(JobConf job, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapred.RangeInputSplit split) {
+ protected void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split) {
--- End diff --
shouldn't the method being deleted be deprecated?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by wjsl <gi...@git.apache.org>.
Github user wjsl commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91043317
> It seems like this change could encourage users to pass many ranges as configuration for the map reduce job.
The current InputFormat already does this, so I don't think it's a big deal that this patch doesn't address this. Also, you have to stuff the input information in the Configuration any ways so that the client can create all of the input splits (and thus store them in memory) before submitting the jrrb.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91240448
Another possible option for supporting a large number of ranges is to add a method called `setRangesFile(conf, hdfsRangesFile)`. A method like this could be added later and would not change any of the API added in this PR, i think it would only change implementation. I think this can achieve the same goal I was trying to achieve with the functor approach. I think its easier to understand and can be easily added after these changes are released.
It is similar to [ACCUMULO-507](https://issues.apache.org/jira/browse/ACCUMULO-507), but avoids the problem of Accumulo code creating a file that can not be reliably cleaned up. The user creates the file, knows its exists, and can clean it up.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27628588
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -458,53 +512,27 @@ public void initialize(InputSplit inSplit, JobConf job) throws IOException {
columns = tableConfig.getFetchedColumns();
}
- try {
- log.debug("Creating connector with user: " + principal);
- log.debug("Creating scanner for table: " + table);
- log.debug("Authorizations are: " + authorizations);
- if (isOffline) {
- scanner = new OfflineScanner(instance, new Credentials(principal, token), split.getTableId(), authorizations);
- } else if (instance instanceof MockInstance) {
- scanner = instance.getConnector(principal, token).createScanner(split.getTableName(), authorizations);
- } else {
- ClientConfiguration clientConf = getClientConfiguration(job);
- ClientContext context = new ClientContext(instance, new Credentials(principal, token), clientConf);
- scanner = new ScannerImpl(context, split.getTableId(), authorizations);
- }
- if (isIsolated) {
- log.info("Creating isolated scanner");
- scanner = new IsolatedScanner(scanner);
- }
- if (usesLocalIterators) {
- log.info("Using local iterators");
- scanner = new ClientSideIteratorScanner(scanner);
- }
- setupIterators(job, scanner, split.getTableName(), split);
- } catch (Exception e) {
- throw new IOException(e);
- }
-
// setup a scanner within the bounds of this split
- for (Pair<Text,Text> c : columns) {
+ for (Pair<Text, Text> c : columns) {
--- End diff --
Extra unnecessary space.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by ctubbsii <gi...@git.apache.org>.
Github user ctubbsii commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91053729
> Gentle reminder that you're looking a gift horse in the mouth with this.
Perhaps, but isn't that the purpose of a code review? To my knowledge, nobody has expressed any ill-will or animosity. I, for one, am appreciative that somebody has taken the time to investigate this. Comprehensive feedback and discussion, and alternative proposals, should not be construed as judgmental or lacking gratitude, but simply as attempts to flesh out the eventual solution to be merged.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28597050
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
I agree, that case doesn't make much sense. I'll add an exception and mention it in javadoc.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27918917
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan && supportBatchScan) {
--- End diff --
> Mostly for code clarity
It threw me a bit, made me wonder if there was a code path I was missing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by ctubbsii <gi...@git.apache.org>.
Github user ctubbsii commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91037784
In JIRA, I [mentioned][1] "sometimes it's better to query a larger range and let an iterator filter out non-matching results".
I think the createRanges method @keith-turner describes could work if the function is executed in the RecordReader (it also simplifies this issue significantly, because you wouldn't need to create a new InputSplit type, but simply add an option to the AccumuloInputFormat). There's still some risk of memory exhaustion with a large number of ranges within a tablet (especially if the ranges were an exhaustive set of row-records to retrieve).
However, I still think that for many things, it's probably better to simply use an iterator with some filter criteria. It could be a SkippingIterator that seeks to ranges which are pre-configured on that iterator, or it could be a Filter which has some filter criteria.
[1]: https://issues.apache.org/jira/browse/ACCUMULO-3602?focusedCommentId=14327767&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-14327767
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91054546
> Eugene has done a lot of work here as a first contribution.
I think the work @echeipesh has done for this is spectacular. I hope my discussion does not discourage. I wish these top level comments on GH could be threaded.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-94077039
`mvn verify -Psunny` came back positive locally. Committed as 956a50e. Thanks for all of your hard work, @echeipesh!!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91641526
@keith-turner I don't think the range decomposition interface approach would handle my case without some kind of intermediate storage level, as I am reading it anyway. The Range -> List<Range> interface does not have enough parameters to determine the query plan. Also, if the query is user generated then a named class can not be present to encapsulate the parameters.
From this standpoint it seems there are two options
- Use some presence that is that would be populated by the query planner
- Perhaps the Configuration object would still need a value to find the correct location
- Expand the interface to to include some kind of `QueryConfiguration` interface that could be implemented by the user and would be passed in.
The second approach sounds more interesting but it's hard to judge it since I don't think I have that problem.
To clarify, what is the underlying concern, that the Configuration object gets too large and the (de)serialization becomes too expensive or not even possible?
My particular use case is reflection on the fact that small ranges introduce too much overhead when they translate to mappers (in MR) and partitions (in spark). So it would be nice if there is some kind of minimum batch, tablet maps nicely to that.
The only other addition to this problem, which is not explored in this PR, is to divide sequence of ranges per tablet server into "balanced", user configurable, number, of groups. Balanced in this case could mean something like: approximately equal number of possible row results. The idea would to be to match "x mappers per host" design. Sounds like it would be great, but I haven't tested anything like that yet, could be another PR if it works out?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28560973
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -296,6 +296,43 @@ protected static boolean isOfflineScan(JobContext context) {
}
/**
+ * Controls the use of the {@link org.apache.accumulo.core.client.BatchScanner} in this job.
--- End diff --
yeah the javadocs are nice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27628943
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/AccumuloInputSplit.java ---
@@ -0,0 +1,425 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.core.client.mapreduce;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.accumulo.core.client.ClientConfiguration;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.IteratorSetting;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.mapreduce.lib.impl.ConfiguratorBase.TokenSource;
+import org.apache.accumulo.core.client.mapreduce.lib.impl.InputConfigurator;
+import org.apache.accumulo.core.client.mock.MockInstance;
+import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
+import org.apache.accumulo.core.client.security.tokens.AuthenticationToken.AuthenticationTokenSerializer;
+import org.apache.accumulo.core.data.ByteSequence;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.util.Base64;
+import org.apache.accumulo.core.util.Pair;
+import org.apache.accumulo.core.data.Key;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.log4j.Level;
+
+/**
+ * Abstracts over configurations common to all InputSplits. Specifically it leaves out methods
+ * related to number of ranges and locations per InputSplit as those vary by implementation.
+ *
+ * @see RangeInputSplit
+ * @see BatchInputSplit
+ */
+public abstract class AccumuloInputSplit extends InputSplit implements Writable {
+ protected String[] locations;
+ protected String tableId, tableName, instanceName, zooKeepers, principal;
+ protected TokenSource tokenSource;
+ protected String tokenFile;
+ protected AuthenticationToken token;
+ protected Boolean mockInstance;
+ protected Authorizations auths;
+ protected Set<Pair<Text,Text>> fetchedColumns;
+ protected List<IteratorSetting> iterators;
+ protected Level level;
+
+ public abstract float getProgress(Key currentKey);
+
+ public AccumuloInputSplit() {
+ locations = new String[0];
+ tableName = "";
+ tableId = "";
+ }
+
+ public AccumuloInputSplit(AccumuloInputSplit split) throws IOException {
+ this.setLocations(split.getLocations());
+ this.setTableName(split.getTableName());
+ this.setTableId(split.getTableId());
+ }
+
+ protected AccumuloInputSplit(String table, String tableId, String[] locations) {
+ setLocations(locations);
+ this.tableName = table;
+ this.tableId = tableId;
+ }
+
+ private static byte[] extractBytes(ByteSequence seq, int numBytes) {
+ byte[] bytes = new byte[numBytes + 1];
+ bytes[0] = 0;
+ for (int i = 0; i < numBytes; i++) {
+ if (i >= seq.length())
+ bytes[i + 1] = 0;
+ else
+ bytes[i + 1] = seq.byteAt(i);
+ }
+ return bytes;
+ }
+
+ public static float getProgress(ByteSequence start, ByteSequence end, ByteSequence position) {
+ int maxDepth = Math.min(Math.max(end.length(), start.length()), position.length());
+ BigInteger startBI = new BigInteger(extractBytes(start, maxDepth));
+ BigInteger endBI = new BigInteger(extractBytes(end, maxDepth));
+ BigInteger positionBI = new BigInteger(extractBytes(position, maxDepth));
+ return (float) (positionBI.subtract(startBI).doubleValue() / endBI.subtract(startBI).doubleValue());
+ }
+
+ public long getRangeLength(Range range) throws IOException {
+ Text startRow = range.isInfiniteStartKey() ? new Text(new byte[] {Byte.MIN_VALUE}) : range.getStartKey().getRow();
+ Text stopRow = range.isInfiniteStopKey() ? new Text(new byte[] {Byte.MAX_VALUE}) : range.getEndKey().getRow();
+ int maxCommon = Math.min(7, Math.min(startRow.getLength(), stopRow.getLength()));
+ long diff = 0;
+
+ byte[] start = startRow.getBytes();
+ byte[] stop = stopRow.getBytes();
+ for (int i = 0; i < maxCommon; ++i) {
+ diff |= 0xff & (start[i] ^ stop[i]);
+ diff <<= Byte.SIZE;
+ }
+
+ if (startRow.getLength() != stopRow.getLength())
+ diff |= 0xff;
+
+ return diff + 1;
+ }
+
+ @Override
+ public String[] getLocations() throws IOException {
+ return Arrays.copyOf(locations, locations.length);
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ tableName = in.readUTF();
+ tableId = in.readUTF();
+ int numLocs = in.readInt();
+ locations = new String[numLocs];
+ for (int i = 0; i < numLocs; ++i)
+ locations[i] = in.readUTF();
+
+ if (in.readBoolean()) {
+ mockInstance = in.readBoolean();
+ }
+
+ if (in.readBoolean()) {
+ int numColumns = in.readInt();
+ List<String> columns = new ArrayList<String>(numColumns);
+ for (int i = 0; i < numColumns; i++) {
+ columns.add(in.readUTF());
+ }
+
+ fetchedColumns = InputConfigurator.deserializeFetchedColumns(columns);
+ }
+
+ if (in.readBoolean()) {
+ String strAuths = in.readUTF();
+ auths = new Authorizations(strAuths.getBytes(UTF_8));
+ }
+
+ if (in.readBoolean()) {
+ principal = in.readUTF();
+ }
+
+ if (in.readBoolean()) {
+ int ordinal = in.readInt();
+ this.tokenSource = TokenSource.values()[ordinal];
+
+ switch (this.tokenSource) {
+ case INLINE:
+ String tokenClass = in.readUTF();
+ byte[] base64TokenBytes = in.readUTF().getBytes(UTF_8);
+ byte[] tokenBytes = Base64.decodeBase64(base64TokenBytes);
+
+ this.token = AuthenticationTokenSerializer.deserialize(tokenClass, tokenBytes);
+ break;
+
+ case FILE:
+ this.tokenFile = in.readUTF();
+
+ break;
+ default:
+ throw new IOException("Cannot parse unknown TokenSource ordinal");
+ }
+ }
+
+ if (in.readBoolean()) {
+ instanceName = in.readUTF();
+ }
+
+ if (in.readBoolean()) {
+ zooKeepers = in.readUTF();
+ }
+
+ if (in.readBoolean()) {
+ int numIterators = in.readInt();
+ iterators = new ArrayList<IteratorSetting>(numIterators);
+ for (int i = 0; i < numIterators; i++) {
+ iterators.add(new IteratorSetting(in));
+ }
+ }
+
+ if (in.readBoolean()) {
+ level = Level.toLevel(in.readInt());
+ }
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ out.writeUTF(tableName);
+ out.writeUTF(tableId);
+ out.writeInt(locations.length);
+ for (int i = 0; i < locations.length; ++i)
+ out.writeUTF(locations[i]);
+
+ out.writeBoolean(null != mockInstance);
+ if (null != mockInstance) {
+ out.writeBoolean(mockInstance);
+ }
+
+ out.writeBoolean(null != fetchedColumns);
+ if (null != fetchedColumns) {
+ String[] cols = InputConfigurator.serializeColumns(fetchedColumns);
+ out.writeInt(cols.length);
+ for (String col : cols) {
+ out.writeUTF(col);
+ }
+ }
+
+ out.writeBoolean(null != auths);
+ if (null != auths) {
+ out.writeUTF(auths.serialize());
+ }
+
+ out.writeBoolean(null != principal);
+ if (null != principal) {
+ out.writeUTF(principal);
+ }
+
+ out.writeBoolean(null != tokenSource);
+ if (null != tokenSource) {
+ out.writeInt(tokenSource.ordinal());
+
+ if (null != token && null != tokenFile) {
+ throw new IOException("Cannot use both inline AuthenticationToken and file-based AuthenticationToken");
+ } else if (null != token) {
+ out.writeUTF(token.getClass().getCanonicalName());
+ out.writeUTF(Base64.encodeBase64String(AuthenticationTokenSerializer.serialize(token)));
+ } else {
+ out.writeUTF(tokenFile);
+ }
+ }
+
+ out.writeBoolean(null != instanceName);
+ if (null != instanceName) {
+ out.writeUTF(instanceName);
+ }
+
+ out.writeBoolean(null != zooKeepers);
+ if (null != zooKeepers) {
+ out.writeUTF(zooKeepers);
+ }
+
+ out.writeBoolean(null != iterators);
+ if (null != iterators) {
+ out.writeInt(iterators.size());
+ for (IteratorSetting iterator : iterators) {
+ iterator.write(out);
+ }
+ }
+
+ out.writeBoolean(null != level);
+ if (null != level) {
+ out.writeInt(level.toInt());
+ }
+ }
+
+ /**
+ * Use {@link #getTableName}
+ */
+ @Deprecated
+ public String getTable() {
+ return getTableName();
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ /**
+ * Use {@link #setTableName}
+ */
+ @Deprecated
+ public void setTable(String table) {
+ setTableName(table);
+ }
+
+ public void setTableName(String table) {
+ this.tableName = table;
+ }
+
+ public void setTableId(String tableId) {
+ this.tableId = tableId;
+ }
+
+ public String getTableId() {
+ return tableId;
+ }
+
+ /**
+ * @see #getInstance(ClientConfiguration)
+ */
+ @Deprecated
--- End diff --
Thanks for catching this already!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91054685
> isn't that the purpose of a code review
Not trying to insinuate there was any ill-will, but this discussion has not been centered around encouragement and has really derailed from the initial changeset. If there is a technical reason why these changes are technically insufficient, then by all means, let's get them addressed. Otherwise, alternative implementations can be done as follow on work (to try to keep Eugene coming back again).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27974998
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -397,15 +414,17 @@ public static InputTableConfig getInputTableConfig(JobConf job, String tableName
* the table name for which the scanner is configured
* @since 1.6.0
*/
- protected abstract void setupIterators(JobConf job, Scanner scanner, String tableName, RangeInputSplit split);
+ @Deprecated
--- End diff --
When deprecated, in addition to adding the annotation we usually add something like `@deprecated since 1.7.0` to the javadoc also. That documents when it was deprecated, I don't think it really serves any other purpose since we have the annotation.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27629593
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -296,6 +296,65 @@ protected static boolean isOfflineScan(JobContext context) {
}
/**
+ * Controls the use of the {@link org.apache.accumulo.core.client.BatchScanner} in this job.
+ * Using this feature will group ranges by their source tablet per InputSplit and use BatchScanner to read them.
+ *
+ * <p>
+ * By default, this feature is <b>disabled</b>.
+ *
+ * @param job
+ * the Hadoop job instance to be configured
+ * @param enableFeature
+ * the feature is enabled if true, disabled otherwise
+ * @since 1.7.0
+ */
+ public static void setBatchScan(Job job, boolean enableFeature) {
+ InputConfigurator.setBatchScan(CLASS, job.getConfiguration(), enableFeature);
+ }
+
+ /**
+ * Determines whether a configuration has the {@link org.apache.accumulo.core.client.BatchScanner} feature enabled.
+ *
+ * @param context
+ * the Hadoop context for the configured job
+ * @since 1.7.0
+ * @see #setBatchScan(Job, boolean)
+ */
+ protected static boolean isBatchScan(JobContext context) {
--- End diff --
Why JobContext here? And why not public?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91052076
> That's orthogonal to accepting arbitrary ranges
I suppose the worst case for the function approach, would be that the configuration for the function is the set of all ranges for all tablets. I am thinking in this case the function proposal would be worse in memory usage than what the PR currently implements. I was thinking the function approach could accomplish anything that the current implementation could, and do some things more efficiently. But now I am not so sure.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28554520
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
I think your intent is fine, it just sounds like we don't want to support `autoAdjust=true` when `batchScan=true`. I think that case is just nonsense and we could catch it in the earlier sanity check where you do the offline/isolate scan or localIterators check. Does that make sense?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27989335
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -384,7 +387,21 @@ public static InputTableConfig getInputTableConfig(JobConf job, String tableName
protected abstract static class AbstractRecordReader<K,V> implements RecordReader<K,V> {
protected long numKeysRead;
protected Iterator<Map.Entry<Key,Value>> scannerIterator;
- protected RangeInputSplit split;
+ protected org.apache.accumulo.core.client.mapreduce.impl.AccumuloInputSplit split;
+ protected ScannerBase scannerBase;
+
+ /**
+ * Configures the iterators on a scanner for the given table name.
+ *
+ * @param job
+ * the Hadoop job configuration
+ * @param scanner
+ * the scanner for which to configure the iterators
+ * @param tableName
+ * the table name for which the scanner is configured
+ * @since 1.7.0
+ */
+ protected abstract void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split);
--- End diff --
> Actually that's just factored out of old RangeInputSplit.
Thanks for the background. I'll take a look at the history of `RangeInputSplit` and see if I can gain any understanding.
The reason I started wondering about this was the following code from your suggestion. I was trying to understand the case when `null == iterators` is true.
```java
iterators = split.getIterators();
if (null == iterators) {
iterators = contextIterators(context, tableName);
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91049787
> How do these changes preclude you from later adding in your proposed split generation code?
They do not. I am just thinking about maintenance over time. If the approach of passing a function is a better solution to the problem at hand, then only we only implement it and have less code to maintain.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-88596530
Ack, I'm sorry, @echeipesh, I just broke you. `RangeInputSplit.getInstance()` is nonsensical. It should use the ClientConfiguration from the Job's configuration (which can only be gotten by the RecordReader using the TaskAttemptContext). LMK if that doesn't make sense and I can send you a patch to fix your branch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91046228
> Also, you have to stuff the input information in the Configuration any ways so that the client can create all of the input splits (and thus store them in memory) before submitting the jrrb.
True in some cases. I was thinking in many cases that a user provided function could deterministically generate a large number of ranges from a small amount of configuration.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91035731
> It seems like this change could encourage users to pass many ranges as configuration for the map reduce job. This could cause memory exhaustion for the job tracker.
Agreed, but is this a new issue? If anything, it this these changes would reduce the total number of objects in memory (as a split could contain multiple ranges whereas previously it was 1:1). I remember hearing about solutions that use HDFS as a mechanism for supporting large numbers of ranges instead of the Configuration. Either way, I think the current changes are a good addition but just not in scope to address the bigger issue you brought up.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27926355
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -629,32 +676,37 @@ public float getProgress() throws IOException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
--- End diff --
Might this need to adhere to autoAdjust for correctness? I think it might be as simple as not clipping the ranges when autoAdjust is false.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28016992
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -384,7 +387,21 @@ public static InputTableConfig getInputTableConfig(JobConf job, String tableName
protected abstract static class AbstractRecordReader<K,V> implements RecordReader<K,V> {
protected long numKeysRead;
protected Iterator<Map.Entry<Key,Value>> scannerIterator;
- protected RangeInputSplit split;
+ protected org.apache.accumulo.core.client.mapreduce.impl.AccumuloInputSplit split;
+ protected ScannerBase scannerBase;
+
+ /**
+ * Configures the iterators on a scanner for the given table name.
+ *
+ * @param job
+ * the Hadoop job configuration
+ * @param scanner
+ * the scanner for which to configure the iterators
+ * @param tableName
+ * the table name for which the scanner is configured
+ * @since 1.7.0
+ */
+ protected abstract void setupIterators(JobConf job, ScannerBase scanner, String tableName, AccumuloInputSplit split);
--- End diff --
> Looking at code right now the only part that diverges is pulling out default split if they're somehow unavailable in the split. This narrows the interface, hopefully in an acceptable way?
@echeipesh I think your approach looks good.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27826065
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java ---
@@ -426,6 +442,7 @@ protected static ClientConfiguration getClientConfiguration(JobContext context)
* the table name for which the scanner is configured
* @since 1.6.0
*/
+ @Deprecated
protected abstract void setupIterators(TaskAttemptContext context, Scanner scanner, String tableName, RangeInputSplit split);
--- End diff --
Could this be implemented in terms of new setupIterators without breaking semver? Would save some ugliness.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27629581
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/BatchInputSplit.java ---
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.core.client.mapreduce;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+
+/**
+ * The Class BatchInputSplit. Encapsulates a set of Accumulo ranges on a single tablet for use in Map Reduce jobs.
+ * Can contain several Ranges per split.
+ */
+public class BatchInputSplit extends AccumuloInputSplit {
+ private Collection<Range> ranges;
+ private int scanThreads;
+
+ public BatchInputSplit() {
+ ranges = Collections.emptyList();
+ }
+
+ public BatchInputSplit(BatchInputSplit split) throws IOException {
+ super(split);
+ this.setRanges(split.getRanges());
+ }
+
+ protected BatchInputSplit(String table, String tableId, Collection<Range> ranges, String[] locations) {
+ super(table, tableId, locations);
+ this.ranges = ranges;
+ }
+
+ public float getProgress(Key currentKey) {
+ if (currentKey == null)
+ return 0f;
+ for (Range range : ranges) {
--- End diff --
Of course, then I think of something. Perhaps keep a sum of the progress over each Range and then return the average across the progress of each Range. It's not guaranteed to be right since we may never get this method called with each Range, but it might be close enough. The important part is ensuring that this number will be increasing.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27629523
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/BatchInputSplit.java ---
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.core.client.mapreduce;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+
+/**
+ * The Class BatchInputSplit. Encapsulates a set of Accumulo ranges on a single tablet for use in Map Reduce jobs.
+ * Can contain several Ranges per split.
+ */
+public class BatchInputSplit extends AccumuloInputSplit {
+ private Collection<Range> ranges;
+ private int scanThreads;
+
+ public BatchInputSplit() {
+ ranges = Collections.emptyList();
+ }
+
+ public BatchInputSplit(BatchInputSplit split) throws IOException {
+ super(split);
+ this.setRanges(split.getRanges());
+ }
+
+ protected BatchInputSplit(String table, String tableId, Collection<Range> ranges, String[] locations) {
+ super(table, tableId, locations);
+ this.ranges = ranges;
+ }
+
+ public float getProgress(Key currentKey) {
+ if (currentKey == null)
+ return 0f;
+ for (Range range : ranges) {
--- End diff --
Sadly, this isn't really representative like it was for a single Range per InputSplit. There will be a semblance of ordering (as Accumulo is going to be doing batches under the hood), but there's no real progress you can identify out of keys from a BatchScanner. The only realistic approximation I can think of is based off of some stats we store in the metadata table, but I don't think that's realistic in this changeset.
I can't think of a reliable way to do this computation. If that's the case, I'd be apt to just `return 0` and mark it as a TODO
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91656749
> To clarify, what is the underlying concern, that the Configuration object gets too large and the (de)serialization becomes too expensive or not even possible?
Yeah that was my concern. I am also making the assumption that the Configuration has to be able to fit into memory on the client submitting the M/R job and on the jobtracker.
> My particular use case is reflection on the fact that small ranges introduce too much overhead when they translate to mappers (in MR) and partitions (in spark). So it would be nice if there is some kind of minimum batch, tablet maps nicely to that.
I think what you are doing to make AIF handle this case well is useful. I have had multiple users ask about this type of thing before. Your changes make it possible for a map reduce job to handle a large number of ranges in an efficient manner, but a user can only pass a set of ranges that fits in memory. I don't think that problem has to be solved in this PR, but I think its useful to think ahead w.r.t to the API. After this change it would be cool if we were not limited by the number of ranges that could fit into memory.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28554794
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/InputFormatBase.java ---
@@ -315,19 +352,8 @@ protected static TabletLocator getTabletLocator(JobConf job) throws TableNotFoun
protected abstract static class RecordReaderBase<K,V> extends AbstractRecordReader<K,V> {
@Override
- protected void setupIterators(JobConf job, Scanner scanner, String tableName, org.apache.accumulo.core.client.mapred.RangeInputSplit split) {
- List<IteratorSetting> iterators = null;
-
- if (null == split) {
- iterators = getIterators(job);
- } else {
- iterators = split.getIterators();
- if (null == iterators) {
- iterators = getIterators(job);
- }
- }
-
- setupIterators(iterators, scanner);
+ protected List<IteratorSetting> jobIterators(JobConf job, String tableName) {
+ return getIterators(job);
--- End diff --
I'm not seeing where this is implemented. Am I just blind?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27628573
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/AbstractInputFormat.java ---
@@ -433,24 +435,76 @@ public void initialize(InputSplit inSplit, JobConf job) throws IOException {
// but the scanner will use the table id resolved at job setup time
InputTableConfig tableConfig = getInputTableConfig(job, split.getTableName());
- Boolean isOffline = split.isOffline();
- if (null == isOffline) {
- isOffline = tableConfig.isOfflineScan();
- }
+ log.debug("Creating connector with user: " + principal);
+ log.debug("Creating scanner for table: " + table);
+ log.debug("Authorizations are: " + authorizations);
- Boolean isIsolated = split.isIsolatedScan();
- if (null == isIsolated) {
- isIsolated = tableConfig.shouldUseIsolatedScanners();
- }
+ if (split instanceof org.apache.accumulo.core.client.mapreduce.RangeInputSplit) {
+ org.apache.accumulo.core.client.mapreduce.RangeInputSplit rangeSplit = (org.apache.accumulo.core.client.mapreduce.RangeInputSplit) split;
- Boolean usesLocalIterators = split.usesLocalIterators();
- if (null == usesLocalIterators) {
- usesLocalIterators = tableConfig.shouldUseLocalIterators();
- }
+ Boolean isOffline = rangeSplit.isOffline();
+ if (null == isOffline) {
+ isOffline = tableConfig.isOfflineScan();
+ }
+
+ Boolean isIsolated = rangeSplit.isIsolatedScan();
+ if (null == isIsolated) {
+ isIsolated = tableConfig.shouldUseIsolatedScanners();
+ }
- List<IteratorSetting> iterators = split.getIterators();
- if (null == iterators) {
- iterators = tableConfig.getIterators();
+ Boolean usesLocalIterators = rangeSplit.usesLocalIterators();
+ if (null == usesLocalIterators) {
+ usesLocalIterators = tableConfig.shouldUseLocalIterators();
+ }
+
+ Scanner scanner;
+
+ try {
+ if (isOffline) {
+ scanner = new OfflineScanner(instance, new Credentials(principal, token), split.getTableId(), authorizations);
+ } else if (instance instanceof MockInstance) {
+ scanner = instance.getConnector(principal, token).createScanner(split.getTableName(), authorizations);
+ } else {
+ ClientConfiguration clientConf = getClientConfiguration(job);
+ ClientContext context = new ClientContext(instance, new Credentials(principal, token), clientConf);
+ scanner = new ScannerImpl(context, split.getTableId(), authorizations);
+ }
+ if (isIsolated) {
+ log.info("Creating isolated scanner");
+ scanner = new IsolatedScanner(scanner);
+ }
+ if (usesLocalIterators) {
+ log.info("Using local iterators");
+ scanner = new ClientSideIteratorScanner(scanner);
+ }
+ setupIterators(job, scanner, split.getTableName(), split);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+
+ scanner.setRange(rangeSplit.getRange());
+
+ // do this last after setting all scanner options
+ scannerIterator = scanner.iterator();
--- End diff --
Duplicated call to `scanner.iterator()`. Looks like you do it down below too?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91049205
> If its not discussed now, then it will likely not be later. 1.7.0 will be released, and we will be supporting these API changes for a long time
How do these changes preclude you from later adding in your proposed split generation code?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27628822
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapred/BatchInputSplit.java ---
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.core.client.mapred;
+
+import java.io.IOException;
+import java.util.Collection;
+
+import org.apache.accumulo.core.data.Range;
+import org.apache.hadoop.mapred.InputSplit;
+
+/**
+ * The Class BatchInputSplit. Encapsulates Accumulo ranges for use in Map Reduce jobs.
+ * Can contain several Ranges per InputSplit.
+ */
+public class BatchInputSplit extends org.apache.accumulo.core.client.mapreduce.BatchInputSplit implements InputSplit {
--- End diff --
I'm not sure if we want these Split classes to actually be public API (https://github.com/apache/accumulo/blob/master/README.md#api). Probably better to put this in an impl/ package. For RangeInputSplit, it kind of got grandfathered in.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-93856225
Ok, two things I think need to be fixed. Everything else you fixed looks good!
1. Restore the `setupIterators(TaskAttemptContext, Scanner, RangeInputSplit)` method (this is another one of those crappy edge cases, sorry).
2. Fail if `batchScan==true` while `autoAdjust==true`
I pulled down the PR and ran it through japi which is how I noticed the first point. Otherwise it looks good from a compatibility sense for 1.6->1.7.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-94019216
+1
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27628899
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/AbstractInputFormat.java ---
@@ -675,32 +711,53 @@ public V getCurrentValue() throws IOException, InterruptedException {
}
for (Map.Entry<KeyExtent,List<Range>> extentRanges : tserverBin.getValue().entrySet()) {
Range ke = extentRanges.getKey().toDataRange();
- for (Range r : extentRanges.getValue()) {
- if (autoAdjust) {
- // divide ranges into smaller ranges, based on the tablets
- RangeInputSplit split = new RangeInputSplit(tableName, tableId, ke.clip(r), new String[] {location});
-
- split.setOffline(tableConfig.isOfflineScan());
- split.setIsolatedScan(tableConfig.shouldUseIsolatedScanners());
- split.setUsesLocalIterators(tableConfig.shouldUseLocalIterators());
- split.setMockInstance(mockInstance);
- split.setFetchedColumns(tableConfig.getFetchedColumns());
- split.setPrincipal(principal);
- split.setToken(token);
- split.setInstanceName(instance.getInstanceName());
- split.setZooKeepers(instance.getZooKeepers());
- split.setAuths(auths);
- split.setIterators(tableConfig.getIterators());
- split.setLogLevel(logLevel);
-
- splits.add(split);
- } else {
- // don't divide ranges
- ArrayList<String> locations = splitsToAdd.get(r);
- if (locations == null)
- locations = new ArrayList<String>(1);
- locations.add(location);
- splitsToAdd.put(r, locations);
+ if (batchScan && supportBatchScan) {
+ // group ranges by tablet to be read by a BatchScanner
+ ArrayList<Range> clippedRanges = new ArrayList<Range>();
+ for(Range r: extentRanges.getValue())
+ clippedRanges.add(ke.clip(r));
+ BatchInputSplit split = new BatchInputSplit(tableName, tableId, clippedRanges, new String[] {location});
+ split.setMockInstance(mockInstance);
--- End diff --
Same point as before -- would be nice to consolidate these setters where possible.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-93865149
I think this looks good. I am finished looking at it. I also think adding the `autoAdjust` sanity check and stating both are not supported in javadoc would be nice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28599303
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -325,27 +362,12 @@ protected void setupIterators(TaskAttemptContext context, Scanner scanner, Strin
* the Hadoop context for the configured job
* @param scanner
* the scanner to configure
+ * @deprecated since 1.7.0; Use {@link #contextIterators} instead.
*/
@Deprecated
protected void setupIterators(TaskAttemptContext context, Scanner scanner) {
- setupIterators(context, scanner, null);
- }
-
- /**
- * Initialize a scanner over the given input split using this task attempt configuration.
- */
- protected void setupIterators(TaskAttemptContext context, Scanner scanner, org.apache.accumulo.core.client.mapreduce.RangeInputSplit split) {
--- End diff --
`setupIterators` with that sig became concrete in super type, `AbstractInputFormat`
https://github.com/apache/accumulo/pull/25/files#diff-696314c306e81784094e2dc76341728bR479
I understood that would keep the compatability.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by echeipesh <gi...@git.apache.org>.
Github user echeipesh commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-90621057
Not sure why findbugs is unhappy. The tests in AccumuloInputFormatIT are following previously used pattern.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27976100
--- Diff: test/src/test/java/org/apache/accumulo/test/functional/AccumuloInputFormatIT.java ---
@@ -146,6 +146,41 @@ public void testGetSplits() throws IOException, AccumuloSecurityException, Accum
AccumuloInputFormat.setAutoAdjustRanges(job, false);
splits = inputFormat.getSplits(job);
assertEquals(ranges.size(), splits.size());
+
+ //BatchScan not available for offline scans
+ AccumuloInputFormat.setBatchScan(job, true);
+
+ AccumuloInputFormat.setOfflineTableScan(job, true);
+ try {
+ inputFormat.getSplits(job);
+ fail("An exception should have been thrown");
+ } catch (IOException e) {}
+ AccumuloInputFormat.setOfflineTableScan(job, false);
+
+ //BatchScan not available with isolated iterators
+ AccumuloInputFormat.setScanIsolation(job, true);
+ try {
+ inputFormat.getSplits(job);
+ fail("An exception should have been thrown");
+ } catch (IOException e) {}
--- End diff --
couldn't this fail because `setOfflineTableScan()` was called earlier? Therefore if `setScanIsolation()` does not cause a failure, the test may not detect this.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r28601704
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/InputFormatBase.java ---
@@ -325,27 +362,12 @@ protected void setupIterators(TaskAttemptContext context, Scanner scanner, Strin
* the Hadoop context for the configured job
* @param scanner
* the scanner to configure
+ * @deprecated since 1.7.0; Use {@link #contextIterators} instead.
*/
@Deprecated
protected void setupIterators(TaskAttemptContext context, Scanner scanner) {
- setupIterators(context, scanner, null);
- }
-
- /**
- * Initialize a scanner over the given input split using this task attempt configuration.
- */
- protected void setupIterators(TaskAttemptContext context, Scanner scanner, org.apache.accumulo.core.client.mapreduce.RangeInputSplit split) {
--- End diff --
Those are different methods. Note that the one you linked includes a `String tableName` argument where the removed one does not. I think we need both methods.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on a diff in the pull request:
https://github.com/apache/accumulo/pull/25#discussion_r27629755
--- Diff: core/src/main/java/org/apache/accumulo/core/client/mapreduce/RangeInputSplit.java ---
@@ -135,38 +72,15 @@ public float getProgress(Key currentKey) {
*/
@Override
public long getLength() throws IOException {
- Text startRow = range.isInfiniteStartKey() ? new Text(new byte[] {Byte.MIN_VALUE}) : range.getStartKey().getRow();
- Text stopRow = range.isInfiniteStopKey() ? new Text(new byte[] {Byte.MAX_VALUE}) : range.getEndKey().getRow();
- int maxCommon = Math.min(7, Math.min(startRow.getLength(), stopRow.getLength()));
- long diff = 0;
-
- byte[] start = startRow.getBytes();
- byte[] stop = stopRow.getBytes();
- for (int i = 0; i < maxCommon; ++i) {
- diff |= 0xff & (start[i] ^ stop[i]);
- diff <<= Byte.SIZE;
- }
-
- if (startRow.getLength() != stopRow.getLength())
- diff |= 0xff;
-
- return diff + 1;
+ return getRangeLength(range);
}
- @Override
- public String[] getLocations() throws IOException {
- return Arrays.copyOf(locations, locations.length);
- }
@Override
public void readFields(DataInput in) throws IOException {
--- End diff --
Need to get offline from the serialization.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-88709750
Overall, really nice stuff! A little cleanup and I think this would be a very nice addition for 1.7.0.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91047735
> I'm happy to continue the discussion for dealing with large numbers of ranges efficiently, I'm just not convinced this is the place to do it.
I think it is. If its not discussed now, then it will likely not be later. 1.7.0 will be released, and we will be supporting these API changes for a long time. I see no harm discussing the design and intentions behind a PR. @echeipesh I am curious if passing a function w/ config would even work for your use case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by joshelser <gi...@git.apache.org>.
Github user joshelser commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91041003
> > Agreed, but is this a new issue?
> I think the intent of this PR is new. I think the intent is to efficiently support a map reduce job that reads from many small ranges. If we are going to do that, then lets do it in a way that scales better.
Re-reading the issue description, yes, you are correct. However, what Eugene has already done is already an improvement over what currently exists, IMO. Assuming that the majority of Ranges that his Spark workflow creates are non-overlapping (a hunch given the subject matter -- very small boxes of interest), this still reduces the total memory use of splits and Ranges. The worst case would be numerous very large ranges. We're never going to have the ideal solution the first time around -- I just don't want this to stall because we come up with a different solution to an already solved problem.
I'm happy to continue the discussion for dealing with large numbers of ranges efficiently, I'm just not convinced this is the place to do it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91038638
> Agreed, but is this a new issue?
I think the intent of this PR is new. I think the intent is to efficiently support a map reduce job that reads from many small ranges. If we are going to do that, then lets do it in a way that scales better.
> However, I still think that for many things, it's probably better to simply use an iterator with some filter criteria.
This approach would work well for existing input format. The only drawback to this approach is it requires adding custom iterators to tservers classpaths. The batch scanner approach allows efficient reads of many small ranges from a tablet w/o custom iterators. Just need a custom function in the map reduce job to generate the ranges.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] accumulo pull request: ACCUMULO-3602 BatchScanner optimization for...
Posted by keith-turner <gi...@git.apache.org>.
Github user keith-turner commented on the pull request:
https://github.com/apache/accumulo/pull/25#issuecomment-91069467
I was discussing the use case I mentioned offline w/ @ctubbsii. This use case was a large number of ranges that can not be generated by a function. We determined that function could handle this case well by storing the ranges somewhere else beside the job conf. For example could do the following.
* Store 10,000,000 sorted ranges in file in distributed cache (assume thousands of tablets)
* Using the provided function, each mapper opens the file and reads the ranges for the tablet its working on.
* The ranges returned by the function are used to initialize the batch scanner for each mapper.
It seems like all of the use cases that the current implementation satisfies could also be satisfied with the functor implementation.
If this PR does not follow the functor approach discussed, thats ok w/ me. I can open up a follow on issue to record the discussion if its not pursued here.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---