You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@accumulo.apache.org by GitBox <gi...@apache.org> on 2018/09/14 22:25:12 UTC
[GitHub] milleruntime closed pull request #648: Rename new Bulk import API
milleruntime closed pull request #648: Rename new Bulk import API
URL: https://github.com/apache/accumulo/pull/648
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java b/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
index 4de0abe139..e18cc57c3a 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/admin/TableOperations.java
@@ -612,7 +612,7 @@ void setLocalityGroups(String tableName, Map<String,Set<Text>> groups)
* @throws TableNotFoundException
* when the table no longer exists
*
- * @deprecated since 2.0.0 use {@link #addFilesTo(String)} instead.
+ * @deprecated since 2.0.0 use {@link #importDirectory(String)} instead.
*/
@Deprecated
void importDirectory(String tableName, String dir, String failureDir, boolean setTime)
@@ -629,7 +629,7 @@ void importDirectory(String tableName, String dir, String failureDir, boolean se
*
* @see NewTableConfiguration#setTimeType(TimeType)
*/
- ImportDestinationOptions tableTime();
+ ImportMappingOptions tableTime();
/**
* Loads the files into the table.
@@ -639,9 +639,11 @@ void load()
}
/**
+ * Options giving control of how the bulk import file mapping is done.
+ *
* @since 2.0.0
*/
- interface ImportDestinationOptions extends ImportOptions {
+ interface ImportMappingOptions extends ImportOptions {
/**
* This is the default number of threads used to determine where to load files. A suffix of
@@ -662,7 +664,7 @@ void load()
* Files are examined to determine where to load them. This examination is done in the current
* process using multiple threads. If this method is not called, then the client property
* {@code bulk.threads} is used to create a thread pool. This property defaults to
- * {@value ImportDestinationOptions#BULK_LOAD_THREADS_DEFAULT}.
+ * {@value ImportMappingOptions#BULK_LOAD_THREADS_DEFAULT}.
*
* @param service
* Use this executor to run file examination task
@@ -675,7 +677,7 @@ void load()
* Files are examined to determine where to load them. This examination is done in the current
* process using multiple threads. If this method is not called, then the client property
* {@code bulk.threads} is used to create a thread pool. This property defaults to
- * {@value org.apache.accumulo.core.client.admin.TableOperations.ImportDestinationOptions#BULK_LOAD_THREADS_DEFAULT}.
+ * {@value ImportMappingOptions#BULK_LOAD_THREADS_DEFAULT}.
*
* @param numThreads
* Create a thread pool with this many thread to run file examination task.
@@ -686,30 +688,35 @@ void load()
/**
* @since 2.0.0
*/
- interface ImportSourceArguments {
+ interface ImportDestinationArguments {
/**
*
- * @param directory
- * Load files from this directory
+ * @param tableName
+ * Import files to this tableName
*/
- ImportDestinationOptions from(String directory);
+ ImportMappingOptions to(String tableName);
}
/**
* Bulk import the files in a directory into a table. Files can be created using
* {@code AccumuloFileOutputFormat} and {@link RFile#newWriter()}.
- *
* <p>
* This new method of bulk import examines files in the current process outside of holding a table
* lock. The old bulk import method ({@link #importDirectory(String, String, String, boolean)})
* examines files on the server side while holding a table read lock.
- *
* <p>
* This API supports adding files to online and offline tables.
+ * <p>
+ * For example, to bulk import files from the directory 'dir1' into the table 'table1' use the
+ * following code.
+ *
+ * <pre>
+ * client.tableOperations().importDirectory("dir1").to("table1").load();
+ * </pre>
*
* @since 2.0.0
*/
- default ImportSourceArguments addFilesTo(String tableName) {
+ default ImportDestinationArguments importDirectory(String directory) {
throw new UnsupportedOperationException();
}
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/BulkImport.java b/core/src/main/java/org/apache/accumulo/core/client/impl/BulkImport.java
index 7e9544c103..8b6f01afc5 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/BulkImport.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/BulkImport.java
@@ -50,8 +50,8 @@
import org.apache.accumulo.core.client.NamespaceNotFoundException;
import org.apache.accumulo.core.client.TableExistsException;
import org.apache.accumulo.core.client.TableNotFoundException;
-import org.apache.accumulo.core.client.admin.TableOperations.ImportDestinationOptions;
-import org.apache.accumulo.core.client.admin.TableOperations.ImportSourceArguments;
+import org.apache.accumulo.core.client.admin.TableOperations.ImportDestinationArguments;
+import org.apache.accumulo.core.client.admin.TableOperations.ImportMappingOptions;
import org.apache.accumulo.core.client.impl.Bulk.FileInfo;
import org.apache.accumulo.core.client.impl.Bulk.Files;
import org.apache.accumulo.core.client.impl.Table.ID;
@@ -84,27 +84,27 @@
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Sets;
-public class BulkImport implements ImportSourceArguments, ImportDestinationOptions {
+public class BulkImport implements ImportDestinationArguments, ImportMappingOptions {
private static final Logger log = LoggerFactory.getLogger(BulkImport.class);
private boolean setTime = false;
private Executor executor = null;
- private String dir;
+ private final String dir;
private int numThreads = -1;
private final ClientContext context;
- private final String tableName;
+ private String tableName;
private LoadPlan plan = null;
- BulkImport(String tableName, ClientContext context) {
+ BulkImport(String directory, ClientContext context) {
this.context = context;
- this.tableName = Objects.requireNonNull(tableName);
+ this.dir = Objects.requireNonNull(directory);
}
@Override
- public ImportDestinationOptions tableTime() {
+ public ImportMappingOptions tableTime() {
this.setTime = true;
return this;
}
@@ -170,13 +170,13 @@ private Path checkPath(FileSystem fs, String dir) throws IOException, AccumuloEx
}
@Override
- public ImportDestinationOptions executor(Executor service) {
+ public ImportMappingOptions executor(Executor service) {
this.executor = Objects.requireNonNull(service);
return this;
}
@Override
- public ImportDestinationOptions threads(int numThreads) {
+ public ImportMappingOptions threads(int numThreads) {
Preconditions.checkArgument(numThreads > 0, "Non positive number of threads given : %s",
numThreads);
this.numThreads = numThreads;
@@ -184,14 +184,14 @@ public ImportDestinationOptions threads(int numThreads) {
}
@Override
- public ImportDestinationOptions plan(LoadPlan plan) {
+ public ImportMappingOptions plan(LoadPlan plan) {
this.plan = plan;
return this;
}
@Override
- public ImportDestinationOptions from(String directory) {
- this.dir = Objects.requireNonNull(directory);
+ public ImportMappingOptions to(String tableName) {
+ this.tableName = Objects.requireNonNull(tableName);
return this;
}
diff --git a/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java b/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
index 1cbd2ab7dd..82846a2fb7 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/impl/TableOperationsImpl.java
@@ -1949,7 +1949,7 @@ public void removeSummarizers(String tableName, Predicate<SummarizerConfiguratio
}
@Override
- public ImportSourceArguments addFilesTo(String tableName) {
- return new BulkImport(tableName, context);
+ public ImportDestinationArguments importDirectory(String directory) {
+ return new BulkImport(directory, context);
}
}
diff --git a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileWriter.java b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileWriter.java
index e466a09d1e..8b0508853a 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileWriter.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/rfile/RFileWriter.java
@@ -37,7 +37,7 @@
/**
* This class provides an API for writing RFiles. It can be used to create file for bulk import into
- * Accumulo using {@link TableOperations#addFilesTo(String)}
+ * Accumulo using {@link TableOperations#importDirectory(String)}
*
* <p>
* A RFileWriter has the following constraints. Violating these constraints will result in runtime
diff --git a/core/src/main/java/org/apache/accumulo/core/conf/ClientProperty.java b/core/src/main/java/org/apache/accumulo/core/conf/ClientProperty.java
index a5229f6a65..f2fd9d840b 100644
--- a/core/src/main/java/org/apache/accumulo/core/conf/ClientProperty.java
+++ b/core/src/main/java/org/apache/accumulo/core/conf/ClientProperty.java
@@ -25,7 +25,7 @@
import java.util.Properties;
import org.apache.accumulo.core.Constants;
-import org.apache.accumulo.core.client.admin.TableOperations.ImportDestinationOptions;
+import org.apache.accumulo.core.client.admin.TableOperations.ImportMappingOptions;
import org.apache.accumulo.core.client.security.tokens.AuthenticationToken;
import org.apache.accumulo.core.client.security.tokens.CredentialProviderToken;
import org.apache.accumulo.core.client.security.tokens.DelegationToken;
@@ -71,7 +71,7 @@
"Number of concurrent query threads to spawn for querying"),
// Bulk load
- BULK_LOAD_THREADS("bulk.threads", ImportDestinationOptions.BULK_LOAD_THREADS_DEFAULT,
+ BULK_LOAD_THREADS("bulk.threads", ImportMappingOptions.BULK_LOAD_THREADS_DEFAULT,
"The number of threads used to inspect bulk load files to determine where files go. "
+ "If the value ends with C, then it will be multiplied by the number of cores on the "
+ "system. This property is only used by the bulk import API introduced in 2.0.0."),
diff --git a/core/src/main/java/org/apache/accumulo/core/data/LoadPlan.java b/core/src/main/java/org/apache/accumulo/core/data/LoadPlan.java
index 58dfabfd13..be5910a0aa 100644
--- a/core/src/main/java/org/apache/accumulo/core/data/LoadPlan.java
+++ b/core/src/main/java/org/apache/accumulo/core/data/LoadPlan.java
@@ -21,7 +21,7 @@
import java.util.Arrays;
import java.util.Collection;
-import org.apache.accumulo.core.client.admin.TableOperations.ImportDestinationOptions;
+import org.apache.accumulo.core.client.admin.TableOperations.ImportMappingOptions;
import org.apache.hadoop.io.Text;
import com.google.common.base.Preconditions;
@@ -31,7 +31,7 @@
/**
* Information about where to load files into an Accumulo table.
*
- * @see ImportDestinationOptions#plan(LoadPlan)
+ * @see ImportMappingOptions#plan(LoadPlan)
* @since 2.0.0
*/
public class LoadPlan {
diff --git a/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java b/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
index 9a7caa7334..7f964cbd68 100644
--- a/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
+++ b/proxy/src/main/java/org/apache/accumulo/proxy/ProxyServer.java
@@ -59,7 +59,7 @@
import org.apache.accumulo.core.client.admin.ActiveScan;
import org.apache.accumulo.core.client.admin.CompactionConfig;
import org.apache.accumulo.core.client.admin.NewTableConfiguration;
-import org.apache.accumulo.core.client.admin.TableOperations.ImportDestinationOptions;
+import org.apache.accumulo.core.client.admin.TableOperations.ImportMappingOptions;
import org.apache.accumulo.core.client.admin.TimeType;
import org.apache.accumulo.core.client.impl.ClientConfConverter;
import org.apache.accumulo.core.client.impl.Credentials;
@@ -1765,8 +1765,8 @@ public void importDirectory(ByteBuffer login, String tableName, String importDir
org.apache.accumulo.proxy.thrift.AccumuloException,
org.apache.accumulo.proxy.thrift.AccumuloSecurityException, TException {
try {
- ImportDestinationOptions loader = getConnector(login).tableOperations().addFilesTo(tableName)
- .from(importDir);
+ ImportMappingOptions loader = getConnector(login).tableOperations().importDirectory(importDir)
+ .to(tableName);
if (setTime) {
loader.tableTime().load();
} else {
diff --git a/shell/src/main/java/org/apache/accumulo/shell/commands/ImportDirectoryCommand.java b/shell/src/main/java/org/apache/accumulo/shell/commands/ImportDirectoryCommand.java
index 9b01c761cb..6727c28ffb 100644
--- a/shell/src/main/java/org/apache/accumulo/shell/commands/ImportDirectoryCommand.java
+++ b/shell/src/main/java/org/apache/accumulo/shell/commands/ImportDirectoryCommand.java
@@ -49,8 +49,8 @@ public int execute(final String fullCommand, final CommandLine cl, final Shell s
// new bulk import only takes 2 args
if (args.length == 2) {
setTime = Boolean.parseBoolean(cl.getArgs()[1]);
- TableOperations.ImportDestinationOptions bulk = shellState.getConnector().tableOperations()
- .addFilesTo(shellState.getTableName()).from(dir);
+ TableOperations.ImportMappingOptions bulk = shellState.getConnector().tableOperations()
+ .importDirectory(dir).to(shellState.getTableName());
if (setTime)
bulk.tableTime().load();
else
diff --git a/test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java b/test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java
index b98dda0e82..12b725998e 100644
--- a/test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/AuditMessageIT.java
@@ -334,8 +334,8 @@ public void testImportExportOperationsAudits()
// Now do a Directory (bulk) import of the same data.
auditConnector.tableOperations().create(THIRD_TEST_TABLE_NAME);
- auditConnector.tableOperations().addFilesTo(THIRD_TEST_TABLE_NAME)
- .from(exportDirBulk.toString()).load();
+ auditConnector.tableOperations().importDirectory(exportDirBulk.toString())
+ .to(THIRD_TEST_TABLE_NAME).load();
auditConnector.tableOperations().online(OLD_TEST_TABLE_NAME);
// Stop testing activities here
diff --git a/test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java b/test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java
index 0ad0336399..dc3060964e 100644
--- a/test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/BulkImportSequentialRowsIT.java
@@ -96,7 +96,7 @@ public void testBulkImportFailure() throws Exception {
// Then import a single rfile to all the tablets, hoping that we get a failure to import because
// of the balancer moving tablets around
// and then we get to verify that the bug is actually fixed.
- to.addFilesTo(tableName).from(bulk.toString()).load();
+ to.importDirectory(bulk.toString()).to(tableName).load();
// The bug is that some tablets don't get imported into.
assertEquals(NR * NV,
diff --git a/test/src/main/java/org/apache/accumulo/test/NamespacesIT.java b/test/src/main/java/org/apache/accumulo/test/NamespacesIT.java
index 246007164a..6f59631938 100644
--- a/test/src/main/java/org/apache/accumulo/test/NamespacesIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/NamespacesIT.java
@@ -1245,7 +1245,7 @@ public void verifyTableOperationsExceptions() throws Exception {
fail();
break;
case 18:
- ops.addFilesTo(tableName).from("").load();
+ ops.importDirectory("").to(tableName).load();
fail();
break;
case 19:
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkFileIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkFileIT.java
index 05ba141ea9..6b8c646dd9 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/BulkFileIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkFileIT.java
@@ -78,7 +78,7 @@ public void testBulkFile() throws Exception {
writeData(conf, aconf, fs, dir, "f2", 334, 999);
writeData(conf, aconf, fs, dir, "f3", 1000, 1999);
- c.tableOperations().addFilesTo(tableName).from(dir).load();
+ c.tableOperations().importDirectory(dir).to(tableName).load();
FunctionalTestUtils.checkRFiles(c, tableName, 6, 6, 1, 1);
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java
index a642198272..5217cfccdb 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkIT.java
@@ -134,7 +134,7 @@ private static void bulkLoad(Connector c, String tableName, Path bulkFailures, P
c.tableOperations().importDirectory(tableName, files.toString(), bulkFailures.toString(),
false);
} else {
- c.tableOperations().addFilesTo(tableName).from(files.toString()).load();
+ c.tableOperations().importDirectory(files.toString()).to(tableName).load();
}
}
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkLoadIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkLoadIT.java
index a115385d59..15dab60fb6 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/BulkLoadIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkLoadIT.java
@@ -123,7 +123,7 @@ private void testSingleTabletSingleFile(boolean offline) throws Exception {
String h1 = writeData(dir + "/f1.", aconf, 0, 332);
- c.tableOperations().addFilesTo(tableName).from(dir).load();
+ c.tableOperations().importDirectory(dir).to(tableName).load();
if (offline)
c.tableOperations().online(tableName);
@@ -153,7 +153,7 @@ private void testSingleTabletSingleFileNoSplits(boolean offline) throws Exceptio
String h1 = writeData(dir + "/f1.", aconf, 0, 333);
- c.tableOperations().addFilesTo(tableName).from(dir).load();
+ c.tableOperations().importDirectory(dir).to(tableName).load();
if (offline)
c.tableOperations().online(tableName);
@@ -185,7 +185,7 @@ public void testBadPermissions() throws Exception {
FsPermission originalPerms = fs.getFileStatus(rFilePath).getPermission();
fs.setPermission(rFilePath, FsPermission.valueOf("----------"));
try {
- c.tableOperations().addFilesTo(tableName).from(dir).load();
+ c.tableOperations().importDirectory(dir).to(tableName).load();
} catch (Exception e) {
Throwable cause = e.getCause();
if (!(cause instanceof FileNotFoundException)
@@ -198,7 +198,7 @@ public void testBadPermissions() throws Exception {
originalPerms = fs.getFileStatus(new Path(dir)).getPermission();
fs.setPermission(new Path(dir), FsPermission.valueOf("dr--r--r--"));
try {
- c.tableOperations().addFilesTo(tableName).from(dir).load();
+ c.tableOperations().importDirectory(dir).to(tableName).load();
} catch (AccumuloException ae) {
if (!(ae.getCause() instanceof FileNotFoundException))
fail("Expected FileNotFoundException but threw " + ae.getCause());
@@ -245,9 +245,9 @@ private void testBulkFile(boolean offline, boolean usePlan) throws Exception {
.loadFileTo("f2.rf", RangeType.TABLE, row(333), row(999))
.loadFileTo("f3.rf", RangeType.FILE, row(1000), row(1499))
.loadFileTo("f4.rf", RangeType.FILE, row(1500), row(1999)).build();
- c.tableOperations().addFilesTo(tableName).from(dir).plan(loadPlan).load();
+ c.tableOperations().importDirectory(dir).to(tableName).plan(loadPlan).load();
} else {
- c.tableOperations().addFilesTo(tableName).from(dir).load();
+ c.tableOperations().importDirectory(dir).to(tableName).load();
}
if (offline)
@@ -292,7 +292,7 @@ public void testBadLoadPlans() throws Exception {
.loadFileTo("f2.rf", RangeType.TABLE, null, row(666))
.loadFileTo("f3.rf", RangeType.TABLE, null, row(666)).build();
try {
- c.tableOperations().addFilesTo(tableName).from(dir).plan(loadPlan).load();
+ c.tableOperations().importDirectory(dir).to(tableName).plan(loadPlan).load();
fail();
} catch (IllegalArgumentException e) {
// ignore
@@ -301,7 +301,7 @@ public void testBadLoadPlans() throws Exception {
// Create a plan with less files than exists in dir
loadPlan = LoadPlan.builder().loadFileTo("f1.rf", RangeType.TABLE, null, row(333)).build();
try {
- c.tableOperations().addFilesTo(tableName).from(dir).plan(loadPlan).load();
+ c.tableOperations().importDirectory(dir).to(tableName).plan(loadPlan).load();
fail();
} catch (IllegalArgumentException e) {
// ignore
@@ -311,7 +311,7 @@ public void testBadLoadPlans() throws Exception {
loadPlan = LoadPlan.builder().loadFileTo("f1.rf", RangeType.TABLE, null, row(555))
.loadFileTo("f2.rf", RangeType.TABLE, null, row(555)).build();
try {
- c.tableOperations().addFilesTo(tableName).from(dir).plan(loadPlan).load();
+ c.tableOperations().importDirectory(dir).to(tableName).plan(loadPlan).load();
fail();
} catch (AccumuloException e) {
// ignore
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/BulkSplitOptimizationIT.java b/test/src/main/java/org/apache/accumulo/test/functional/BulkSplitOptimizationIT.java
index b17d7a8b54..7bf321e232 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/BulkSplitOptimizationIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/BulkSplitOptimizationIT.java
@@ -92,7 +92,7 @@ public void testBulkSplitOptimization() throws Exception {
FileStatus[] stats = fs.listStatus(testDir);
System.out.println("Number of generated files: " + stats.length);
- c.tableOperations().addFilesTo(tableName).from(testDir.toString()).load();
+ c.tableOperations().importDirectory(testDir.toString()).to(tableName).load();
FunctionalTestUtils.checkSplits(c, tableName, 0, 0);
FunctionalTestUtils.checkRFiles(c, tableName, 1, 1, 100, 100);
diff --git a/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java b/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java
index 498997670a..f4a879b154 100644
--- a/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java
+++ b/test/src/main/java/org/apache/accumulo/test/functional/CompactionIT.java
@@ -114,7 +114,7 @@ public void test() throws Exception {
Path testrf = new Path(root, "testrf");
FunctionalTestUtils.createRFiles(c, fs, testrf.toString(), 500000, 59, 4);
- c.tableOperations().addFilesTo(tableName).from(testrf.toString()).load();
+ c.tableOperations().importDirectory(testrf.toString()).to(tableName).load();
int beforeCount = countFiles(c);
final AtomicBoolean fail = new AtomicBoolean(false);
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services