'Version 1.0.1 of the DynamoDB Import Export Tool'

Author: dymaws

README.md

@@ -1,49 +1,44 @@
 # DynamoDB Import Export Tool
+The DynamoDB Import Export Tool is designed to perform parallel scans on the source table, store scan results in a queue, then consume the queue by writing the items asynchronously to a destination table.
+
+## Requirements ##
+* Maven
+* JRE 1.7+
+* Pre-existing source and destination DynamoDB tables
 
 ## Running as an executable
 
-mvn exec:java -Dexec.mainClass="com.amazonaws.dynamodb.bootstrap.CommandLineInterface" \
+1. Build the library:
+
+```
+    mvn install
+```
+
+2. This produces the target jar in the target/ directory, to start the replication process:
+
+java -jar dynamodb-import-export-tool.jar
 
- -Dexec.args=" \
- 
- --destinationEndpoint dynamodb.us-east-1.amazonaws.com \
- 
- --destinationTable nameOfDestinationTable  \
- 
- --sourceEndpoint dynamodb.us-west-1.amazonaws.com \
- 
- --sourceTable nameOfSourceTable \
- 
- --readThroughputRatio .5  \
- 
- --writeThroughputRatio .7 \
- 
- --maxWriteThreads 128 \
- 
- --totalSections 4 \
- 
- --section 3"
+--destinationEndpoint <destination_endpoint> // the DynamoDB endpoint where the destination table is located.
 
-### Description of Arguments
+--destinationTable <destination_table> // the destination table to write to.
 
-destinationEndpoint: The endpoint where the destination table is located
+--sourceEndpoint <source_endpoint> // the endpoint where the source table is located.
 
-destinationTable: The destination table to write to
+--sourceTable <source_table>// the source table to read from.
 
-sourceEndpoint: The endpoint where the source table is located
+--readThroughputRatio <ratio_in_decimal> // the ratio of read throughput to consume from the source table.
 
-sourceTable: The source table to read from
+--writeThroughputRatio <ratio_in_decimal> // the ratio of write throughput to consume from the destination table.
 
-readThroughputRatio: The ratio of read throughput to consume from the source table
+--maxWriteThreads <numWriteThreads> // (Optional, default=128 * Available_Processors) Maximum number of write threads to create.
 
-writeThroughputRatio: The ratio of write throughput to consume from the destination table
+--totalSections <numSections> // (Optional, default=1) Total number of sections to split the bootstrap into. Each application will only scan and write one section.
 
-maxWriteThreads: (Optional, default=128 * Available_Processors) Maximum number of write threads to create
+--section <sectionSequence> // (Optional, default=0) section to read and write. Only will scan this one section of all sections, [0...totalSections-1].
 
-totalSections: (Optional, default=1) Total number of sections to split the bootstrap into. Each application will only scan and write one section
+--consistentScan <boolean> // (Optional, default=false) indicates whether consistent scan should be used when reading from the source table.
 
-section: (Optional, default=0) section to read and write. Only will scan this one section of all sections, [0...totalSections-1]
- 
+> **NOTE**: To split the replication process across multiple machines, simply use the totalSections & section command line arguments, where each machine will run one section out of [0 ... totalSections-1].
 
 ## Using the API
 

pom.xml

@@ -1,11 +1,15 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.amazonaws</groupId>
-    <version>1.0.0</version>
+    <version>1.0.1</version>
     <artifactId>dynamodb-import-export-tool</artifactId>
     <packaging>jar</packaging>
     <name>DynamoDB Import Export Tool</name>
     <url>https://github.com/awslabs/dynamodb-import-export-tool</url>
+    <description>Exports DynamoDB items via parallel scan into a blocking queue, then consumes the queue and import DynamoDB items into a replica table using asynchronous writes.</description>
+    <scm>
+        <url>https://github.com/awslabs/dynamodb-import-export-tool.git</url>
+    </scm>
     <properties>
         <aws.java.sdk.version>1.10.10</aws.java.sdk.version>
         <powermock.version>1.6.2</powermock.version>
@@ -110,7 +114,6 @@
                 </configuration>
                 <version>3.0</version>
             </plugin>
-            
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-gpg-plugin</artifactId>

src/main/java/com/amazonaws/dynamodb/bootstrap/CommandLineInterface.java

@@ -23,13 +23,15 @@
 import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
+import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
 import com.amazonaws.dynamodb.bootstrap.constants.BootstrapConstants;
 import com.amazonaws.dynamodb.bootstrap.exception.NullReadCapacityException;
 import com.amazonaws.dynamodb.bootstrap.exception.SectionOutOfRangeException;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
 import com.amazonaws.services.dynamodbv2.model.TableDescription;
 import com.beust.jcommander.JCommander;
+import com.beust.jcommander.ParameterException;
 
 /**
  * The interface that parses the arguments, and begins to transfer data from one
@@ -53,8 +55,15 @@ public static void main(String[] args) {
         CommandLineArgs params = new CommandLineArgs();
         JCommander cmd = new JCommander(params);
 
-        // parse given arguments
-        cmd.parse(args);
+        try {
+            // parse given arguments
+            cmd.parse(args);
+        } catch (ParameterException e) {
+            LOGGER.error(e);
+            JCommander.getConsole().println(e.getMessage());
+            cmd.usage();
+            System.exit(1);
+        }
 
         // show usage information if help flag exists
         if (params.getHelp()) {
@@ -70,10 +79,13 @@ public static void main(String[] args) {
         final int maxWriteThreads = params.getMaxWriteThreads();
         final boolean consistentScan = params.getConsistentScan();
 
+        final ClientConfiguration sourceConfig = new ClientConfiguration().withMaxConnections(BootstrapConstants.MAX_CONN_SIZE);
+        final ClientConfiguration destinationConfig = new ClientConfiguration().withMaxConnections(BootstrapConstants.MAX_CONN_SIZE);
+
         final AmazonDynamoDBClient sourceClient = new AmazonDynamoDBClient(
-                new DefaultAWSCredentialsProviderChain());
+                new DefaultAWSCredentialsProviderChain(), sourceConfig);
         final AmazonDynamoDBClient destinationClient = new AmazonDynamoDBClient(
-                new DefaultAWSCredentialsProviderChain());
+                new DefaultAWSCredentialsProviderChain(), destinationConfig);
         sourceClient.setEndpoint(sourceEndpoint);
         destinationClient.setEndpoint(destinationEndpoint);
 
@@ -136,7 +148,7 @@ private static double calculateThroughput(
     /**
      * Returns the thread pool for the destination DynamoDB table.
      */
-    public static ExecutorService getDestinationThreadPool(int maxWriteThreads) {
+    private static ExecutorService getDestinationThreadPool(int maxWriteThreads) {
         int corePoolSize = BootstrapConstants.DYNAMODB_CLIENT_EXECUTOR_CORE_POOL_SIZE;
         if (corePoolSize > maxWriteThreads) {
             corePoolSize = maxWriteThreads - 1;
@@ -166,4 +178,4 @@ private static ExecutorService getSourceThreadPool(int numSegments) {
         return exec;
     }
 
-}
+}

src/main/java/com/amazonaws/dynamodb/bootstrap/DynamoDBBootstrapWorker.java

@@ -64,7 +64,7 @@ public DynamoDBBootstrapWorker(AmazonDynamoDBClient client,
         this.totalSections = totalSections;
         this.consistentScan = consistentScan;
 
-        threadPool = exec;
+        super.threadPool = exec;
     }
 
     /**
@@ -91,7 +91,7 @@ public DynamoDBBootstrapWorker(AmazonDynamoDBClient client,
         if (numProcessors > numThreads) {
             numThreads = numProcessors;
         }
-        this.threadPool = Executors.newFixedThreadPool(numThreads);
+        super.threadPool = Executors.newFixedThreadPool(numThreads);
     }
 
     /**
@@ -105,6 +105,7 @@ public void pipe(final AbstractLogConsumer consumer)
 
         final ScanRequest request = new ScanRequest().withTableName(tableName)
                 .withReturnConsumedCapacity(ReturnConsumedCapacity.TOTAL)
+                .withLimit(BootstrapConstants.SCAN_LIMIT)
                 .withConsistentRead(consistentScan);
 
         final ParallelScanExecutor scanService = scanner
@@ -151,4 +152,4 @@ public static int getNumberOfSegments(TableDescription description)
                 Math.ceil(tableSizeInGigabytes) / 10));
     }
 
-}
+}

src/main/java/com/amazonaws/dynamodb/bootstrap/DynamoDBConsumer.java

@@ -50,8 +50,8 @@ public DynamoDBConsumer(AmazonDynamoDBClient client, String tableName,
         this.client = client;
         this.tableName = tableName;
         this.rateLimiter = RateLimiter.create(rateLimit);
-        threadPool = exec;
-        this.exec = new ExecutorCompletionService<Void>(threadPool);
+        super.threadPool = exec;
+        super.exec = new ExecutorCompletionService<Void>(threadPool);
     }
 
     /**

src/main/java/com/amazonaws/dynamodb/bootstrap/DynamoDBConsumerWorker.java

@@ -100,7 +100,7 @@ public List<ConsumedCapacity> runWithBackoff(BatchWriteItemRequest req) {
                         }
                     }
                 }
-            } while (unprocessedItems.get(tableName) != null);
+            } while (unprocessedItems != null && unprocessedItems.get(tableName) != null);
             return consumedCapacities;
         } finally {
             if (interrupted) {

src/main/java/com/amazonaws/dynamodb/bootstrap/DynamoDBTableScan.java

@@ -76,6 +76,7 @@ public ScanRequest copyScanRequest(ScanRequest request) {
                 .withTotalSegments(request.getTotalSegments())
                 .withSegment(request.getSegment())
                 .withReturnConsumedCapacity(request.getReturnConsumedCapacity())
+                .withLimit(request.getLimit())
                 .withConsistentRead(request.getConsistentRead());
     }
-}
+}

src/main/java/com/amazonaws/dynamodb/bootstrap/ScanSegmentWorker.java

@@ -67,8 +67,8 @@ public SegmentedScanResult call() {
             int itemSize = isConsistent ? BootstrapConstants.STRONGLY_CONSISTENT_READ_ITEM_SIZE
                     : BootstrapConstants.EVENTUALLY_CONSISTENT_READ_ITEM_SIZE;
 
-            lastConsumedCapacity = (int) ((result.getScannedCount() / Math.max(1.0, result.getCount()))
-                    * (ItemSizeCalculator.calculateScanResultSizeInBytes(result) / itemSize));
+            lastConsumedCapacity = (result.getScannedCount() / (int) Math.max(1.0, result.getCount()))
+                    * (ItemSizeCalculator.calculateScanResultSizeInBytes(result) / itemSize);
         }
 
         if (result.getLastEvaluatedKey() != null

src/main/java/com/amazonaws/dynamodb/bootstrap/constants/BootstrapConstants.java

@@ -38,7 +38,7 @@ public class BootstrapConstants {
             .getRuntime().availableProcessors() * 128;
 
     /**
-     * Seconds to wait for thread to terminate before retrying or exiting.
+     * Core pool size of a default thread pool.
      */
     public static final int DYNAMODB_CLIENT_EXECUTOR_CORE_POOL_SIZE = Runtime
             .getRuntime().availableProcessors() * 4;
@@ -83,7 +83,7 @@ public class BootstrapConstants {
     /**
      * The size of an empty document in a DynamoDB item.
      */
-    public static int LOGICAL_SIZE_OF_EMPTY_DOCUMENT = 3;
+    public static final int LOGICAL_SIZE_OF_EMPTY_DOCUMENT = 3;
 
     /**
      * Max number of bytes in a DynamoDB number attribute.
@@ -100,4 +100,13 @@ public class BootstrapConstants {
      */
     public static final int EVENTUALLY_CONSISTENT_READ_ITEM_SIZE = 2 * STRONGLY_CONSISTENT_READ_ITEM_SIZE;
 
+    /**
+     * Max scan result size
+     */
+    public static final int SCAN_LIMIT = 1000;
+
+    /**
+     * Max connection size limit
+     */
+    public static final int MAX_CONN_SIZE = 5000;
 }

src/main/resources/log4j.properties

@@ -0,0 +1,9 @@
+log4j.rootLogger=INFO, CONSOLE
+
+log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender
+log4j.appender.CONSOLE.layout=org.apache.log4j.EnhancedPatternLayout
+log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} %-5p %40C - %m%n%throwable
+log4j.appender.CONSOLE.threshold=INFO
+log4j.appender.CONSOLE.filter.1=org.apache.log4j.varia.StringMatchFilter
+log4j.appender.CONSOLE.filter.1.StringToMatch=has no children
+log4j.appender.CONSOLE.filter.1.AcceptOnMatch=false