dynamo

gistfile1.txt

package net.ibogner.dynamo;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProviderChain;
import com.amazonaws.auth.InstanceProfileCredentialsProvider;
import com.amazonaws.auth.profile.ProfileCredentialsProvider;
import com.amazonaws.regions.Region;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClient;
import com.amazonaws.services.dynamodbv2.document.BatchWriteItemOutcome;
import com.amazonaws.services.dynamodbv2.document.DynamoDB;
import com.amazonaws.services.dynamodbv2.document.Item;
import com.amazonaws.services.dynamodbv2.document.Table;
import com.amazonaws.services.dynamodbv2.document.TableWriteItems;
import com.amazonaws.services.dynamodbv2.model.AttributeDefinition;
import com.amazonaws.services.dynamodbv2.model.BatchWriteItemResult;
import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
import com.amazonaws.services.dynamodbv2.model.KeySchemaElement;
import com.amazonaws.services.dynamodbv2.model.KeyType;
import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException;
import com.amazonaws.services.dynamodbv2.model.ScalarAttributeType;
import com.amazonaws.services.dynamodbv2.model.TableDescription;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.joda.time.Instant;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class DynamoTest {

    private static final Logger log = LoggerFactory.getLogger(DynamoTest.class);

    private static final String credentialsProfileName = "ibognerdev";
    private static final Region region = Region.getRegion(Regions.fromName("us-west-2"));
    private static final String tableName = "mbognertest1";
    private static final long numItemsToInsert = 20L;
    private static final long readCapacityUnits = 1L;
    private static final long writeCapacityUnits = 1L;

    private final DynamoDB dynamoDB;

    public DynamoTest() {
        dynamoDB = getDBI();
    }

    private AWSCredentialsProviderChain getCredentialsChain() {
        return new AWSCredentialsProviderChain(new ProfileCredentialsProvider(credentialsProfileName),
                                               new InstanceProfileCredentialsProvider());
    }

    private ClientConfiguration getClientConfiguration() {
        return new ClientConfiguration()
                .withConnectionTimeout(50 * 1000)
                .withConnectionTTL(-1)
                .withGzip(false)
                .withMaxConnections(50)
                .withReaper(true)
                .withSocketTimeout(50 * 1000);
    }

    private DynamoDB getDBI() {
        return new DynamoDB((new AmazonDynamoDBClient(getCredentialsChain(), getClientConfiguration())).<AmazonDynamoDBClient>withRegion(region));
    }

    public void createTable() throws InterruptedException {
        try {
            dynamoDB.getTable(tableName).describe();
            log.info("Table {} already exists. Skipping creation.", tableName);
            return;
        } catch (ResourceNotFoundException e) {
            // Expected if table doesn't exist. Continue...
        }

        List<AttributeDefinition> attributeDefinitions = ImmutableList.of(new AttributeDefinition("id", ScalarAttributeType.N));
        List<KeySchemaElement> keySchema = ImmutableList.of(new KeySchemaElement("id", KeyType.HASH));

        CreateTableRequest request = new CreateTableRequest()
                .withTableName(tableName)
                .withKeySchema(keySchema)
                .withAttributeDefinitions(attributeDefinitions)
                .withProvisionedThroughput(new ProvisionedThroughput(readCapacityUnits, writeCapacityUnits));

        log.info("Creating table {}", tableName);
        Table table = dynamoDB.createTable(request);
        log.info("Table created. Waiting for it to become active...");
        table.waitForActive();
        log.info("Table active.");
    }

    public void insertItemsAdHoc() {
        log.info("About to ad-hoc insert items...");
        Table table = dynamoDB.getTable(tableName);
        for (int ctr = 0; ctr < numItemsToInsert; ++ctr) {
            log.info("Inserting {} of {} items ad-hoc", (ctr + 1), numItemsToInsert);
            table.putItem(new Item()
                          .withPrimaryKey("id", Instant.now().getMillis())
                          .withString("Name", "xyz+" + Instant.now().getMillis())
                          .withNumber("Counter", 100 + ctr));
        }
    }

    public void insertItemsBatch() {
        log.info("About to batch insert items...");
        TableWriteItems tableWriteItems = new TableWriteItems(tableName);

        for (int ctr = 0; ctr < numItemsToInsert; ++ctr) {
            tableWriteItems.addItemToPut(new Item()
                                                 .withPrimaryKey("id", Instant.now().getMillis() + ctr)
                                                 .withString("Name", "xyz+" + Instant.now().getMillis())
                                                 .withNumber("Counter", 200 + ctr));

        }

        log.info("Inserting {} items in batch...", numItemsToInsert);
        BatchWriteItemOutcome batchWriteItemOutcome = dynamoDB.batchWriteItem(tableWriteItems);
        log.info("Completed batch insert", numItemsToInsert);

        BatchWriteItemResult result = batchWriteItemOutcome.getBatchWriteItemResult();
        log.info("Result {}", result.toString());
    }

    public void describeTable() {
        TableDescription description = dynamoDB.getTable(tableName).describe();
        log.info("Table status {}, rows {}, size-in-bytes {}", description.getTableStatus(), description.getItemCount(), description.getTableSizeBytes());
    }

    public void scanTable() {
        Table table = dynamoDB.getTable(tableName);
        int ctr = 1;
        for (Item myItem : table.scan("#ctr >= :val", ImmutableMap.of("#ctr", "Counter"), ImmutableMap.<String, Object>of(":val", 200))) {
            log.info("Scan item {}: {}", ctr++, myItem.toJSON());
        }
    }

    public void deleteTable() throws InterruptedException {
        Table table = dynamoDB.getTable(tableName);
        log.info("Deleting table {}", tableName);
        table.delete();
        log.info("Waiting for delete to complete...");
        table.waitForDelete();
        log.info("Table deleted.");
    }

    public static void main(String[] args) throws InterruptedException {
        DynamoTest test = new DynamoTest();
        test.createTable();
        test.insertItemsAdHoc();
        test.insertItemsBatch();
        test.scanTable();
        test.describeTable();
        test.deleteTable();
    }
}