import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
/**
  * Method to create the specific client to be used
  * @param clientType
  * @param credentials
  * @return
  */
public AmazonDynamoDB getClient(String clientType, AWSCredentials credentials){
  if (clientType.equals("sync"))
    return new AmazonDynamoDBClient(credentials);
  if (clientType.equals("async"))
    return new AmazonDynamoDBAsyncClient(credentials);
  return null;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
DynamoDBStorage(String tableName, String awsAccessKeyId, String awsSecretKey, 
        AmazonDynamoDBClient dynamo, HadoopJobInfo hadoopJobInfo) {
    this.tableName = tableName;
    this.awsAccessKeyId = awsAccessKeyId;
    this.awsSecretKey = awsSecretKey;
    this.dynamo = dynamo;
    this.hadoopJobInfo = hadoopJobInfo;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
private AmazonDynamoDBClient loadDynamoDB() {
    if (this.dynamo == null) {
        this.dynamo = new AmazonDynamoDBClient(new BasicAWSCredentials(
                this.awsAccessKeyId, this.awsSecretKey));
    }
    return this.dynamo;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
@Test
public void testMissingPrimaryKey() throws IOException, InterruptedException {
    String tableName = "mortar_test_foo_table";
    String awsAccessKeyId = "XXXXXXXXXXXXX";
    String awsSecretKey = "YYYYYYYYYYYYYY";
    ResourceSchema schema = 
            new ResourceSchema(Utils.getSchemaFromString("my_field:int"));
    
 // mock dynamo client
    AmazonDynamoDBClient dynamo = mock(AmazonDynamoDBClient.class);
    DescribeTableResult describeResult = new DescribeTableResult()
        .withTable(
            new TableDescription()
                .withProvisionedThroughput(
                    new ProvisionedThroughputDescription().withWriteCapacityUnits(50L))
                .withKeySchema(new KeySchema()
                    .withHashKeyElement(new KeySchemaElement()
                        .withAttributeName("not_the_key_you_will_find")
                        .withAttributeType(ScalarAttributeType.N))));
             
    when(dynamo.describeTable(any(DescribeTableRequest.class))).thenReturn(describeResult);
    DynamoDBStorage storage = 
            new DynamoDBStorage(tableName, awsAccessKeyId, awsSecretKey, dynamo, null);
    try {
        storage.checkSchema(schema);
        Assert.fail("Expected schema validation to fail");
    } catch(IOException e) {
        Assert.assertTrue("Expected " + e.getMessage() + " to contain hash msg", e.getMessage().contains("hash primary key"));
    }
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
/**
 * Creates the metastore table in DynamoDB if it doesn't exist with the configured
 * read and write units.
 * 
 * @param uri
 * @param conf
 * @throws Exception 
 */
@Override
public void initalize(URI uri, Configuration conf) throws Exception {
    scheme = uri.getScheme();
    
    String keyId = conf.get("fs."+uri.getScheme()+".awsAccessKeyId");
    String keySecret = conf.get("fs."+uri.getScheme()+".awsSecretAccessKey");
    
    //An override option for accessing across accounts
    keyId = conf.get("s3mper.override.awsAccessKeyId", keyId);
    keySecret = conf.get("s3mper.override.awsSecretAccessKey", keySecret);
    
    db = new AmazonDynamoDBClient(new BasicAWSCredentials(keyId, keySecret));
    
    readUnits = conf.getLong("s3mper.metastore.read.units", readUnits);
    writeUnits = conf.getLong("s3mper.metastore.write.units", writeUnits);
    
    retryCount = conf.getInt("s3mper.metastore.retry", retryCount);
    timeout = conf.getInt("s3mper.metastore.timeout", timeout);
    
    tableName = conf.get("s3mper.metastore.name", tableName);
    
    deleteMarkerEnabled = conf.getBoolean("s3mper.metastore.deleteMarker.enabled", false);
    
    boolean checkTableExists = conf.getBoolean("s3mper.metastore.create", false);
    
    if(checkTableExists) {
        ListTablesResult tables = db.listTables();
        
        if(!tables.getTableNames().contains(tableName)) {
            createTable();
        }
    }
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
public void initalize(URI uri, Configuration conf) throws Exception {
    String keyId = conf.get("fs."+uri.getScheme()+".awsAccessKeyId");
    String keySecret = conf.get("fs."+uri.getScheme()+".awsSecretAccessKey");
    
    //An override option for accessing across accounts
    keyId = conf.get("s3mper.override.awsAccessKeyId", keyId);
    keySecret = conf.get("s3mper.override.awsSecretAccessKey", keySecret);
    
    db = new AmazonDynamoDBClient(new BasicAWSCredentials(keyId, keySecret));
    
    tableName = conf.get("s3mper.metastore.name", tableName);
    
    metastore = new DynamoDBMetastore();
    metastore.initalize(uri, conf);
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
public TimeseriesScannerTask(AmazonDynamoDBClient db, RateLimiter limiter, BlockingQueue<Key> deleteQueue, int queueSize, long age) {
    super(db, limiter);
    this.deleteQueue = deleteQueue;
    this.queueSize = queueSize;
    this.age = age;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
public AbstractScannerTask(AmazonDynamoDBClient db, RateLimiter limiter) {
    super(db, limiter);
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
public PathScannerTask(AmazonDynamoDBClient db, RateLimiter limiter, BlockingQueue<Key> deleteQueue, int queueSize, long age) {
    super(db, limiter);
    this.deleteQueue = deleteQueue;
    this.age = age;
    this.queueSize = queueSize;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
public DeleteWriterTask(AmazonDynamoDBClient db, RateLimiter limiter, BlockingQueue<Key> deleteQueue) {
    super(db, limiter);
    this.deleteQueue = deleteQueue;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
public AbstractDynamoDBTask(AmazonDynamoDBClient db, RateLimiter limiter) {
    this.db = db;
    this.limiter = limiter;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
public AmazonDynamoDBClient getDb() {
    return db;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
public void setDb(AmazonDynamoDBClient db) {
    this.db = db;
}
 

import com.amazonaws.services.dynamodb.AmazonDynamoDBClient; //导入依赖的package包/类
/**
 * Gets DynamoDBClient to be used
 * @return
 */
public AmazonDynamoDBClient getDynamoDBClient() {
  return dynamoDBClient;
}