diff --git a/dynamodb/README.md b/dynamodb/README.md
index 61455ed919..04ed25ef80 100644
--- a/dynamodb/README.md
+++ b/dynamodb/README.md
@@ -36,7 +36,14 @@ for more information on setup.
 # Properties
 
     $DYNAMODB_HOME/conf/dynamodb.properties
-    $DYNAMODB_HOME/conf/AWSCredentials.properties
+
+# AWS credentials
+
+To make requests to Amazon Web Services using the AWS SDK for Java, you must use 
+cryptographically-signed credentials issued by AWS.
+The DynamoDB Client is using the Default Credential Provider chain by default. The available
+configuration option to authenticate your client correctly are documented here:
+https://docs.aws.amazon.com/sdk-for-java/v2/developer-guide/credentials.html#credentials-default
 
 # FAQs
 * Why is the recommended workload distribution set to 'uniform'?
diff --git a/dynamodb/conf/AWSCredentials.properties b/dynamodb/conf/AWSCredentials.properties
deleted file mode 100644
index e337b391f0..0000000000
--- a/dynamodb/conf/AWSCredentials.properties
+++ /dev/null
@@ -1,19 +0,0 @@
-# Copyright (c) 2012 YCSB contributors. All rights reserved.
-#
-# Licensed 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. See accompanying
-# LICENSE file.
-
-# Fill in your AWS Access Key ID and Secret Access Key
-# http://aws.amazon.com/security-credentials
-#accessKey =
-#secretKey =
diff --git a/dynamodb/conf/dynamodb.properties b/dynamodb/conf/dynamodb.properties
index 666f33dbe1..e52daf41b6 100644
--- a/dynamodb/conf/dynamodb.properties
+++ b/dynamodb/conf/dynamodb.properties
@@ -18,9 +18,6 @@
 
 ## Mandatory parameters
 
-# AWS credentials associated with your aws account.
-#dynamodb.awsCredentialsFile = <path to AWSCredentials.properties>
-
 # Primarykey of table 'usertable'
 #dynamodb.primaryKey = <firstname>
 
@@ -69,7 +66,7 @@
 #dynamodb.hashKeyValue = <some value of your choice>
 
 # AWS Region code to connect to:
-# https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-regions-availability-zones.html#concepts-available-regions 
+# https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-regions-availability-zones.html#concepts-available-regions
 # Set this parameter, unless you are using the default value ('us-east-1).
 #dynamodb.region = us-east-1
 
@@ -86,14 +83,16 @@
 # "true" or "false"
 #dynamodb.debug = false
 
-# Maximum number of concurrent connections
-#dynamodb.connectMax = 50
-
-# Read consistency.Consistent reads are expensive and consume twice 
+# Read consistency. Consistent reads are expensive and consume twice
 # as many resources as eventually consistent reads. Defaults to false.
 # "true" or "false"
 #dynamodb.consistentReads = false
 
+# Batch inserts. Use BatchWriteItem API for loading data into the table more efficiently.
+# Especially relevant during the YCSB load phase.
+# "true" or "false". Defaults to false
+#dynamodb.batchInsert = false
+
 # Workload size has implications on provisioned read and write
 # capacity units.Refer FAQs in README
 #fieldcount = 10
diff --git a/dynamodb/pom.xml b/dynamodb/pom.xml
index cb9a457144..5d560e2b3b 100644
--- a/dynamodb/pom.xml
+++ b/dynamodb/pom.xml
@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<!-- 
+<!--
 Copyright (c) 2012 - 2016 YCSB contributors. All rights reserved.
 
 Licensed under the Apache License, Version 2.0 (the "License"); you
@@ -35,9 +35,14 @@ LICENSE file.
 
   <dependencies>
     <dependency>
-      <groupId>com.amazonaws</groupId>
-      <artifactId>aws-java-sdk</artifactId>
-      <version>1.12.565</version>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>apache-client</artifactId>
+      <version>2.27.21</version>
+    </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>dynamodb</artifactId>
+      <version>2.27.21</version>
     </dependency>
     <dependency>
       <groupId>log4j</groupId>
diff --git a/dynamodb/src/main/java/site/ycsb/db/DynamoDBClient.java b/dynamodb/src/main/java/site/ycsb/db/DynamoDBClient.java
index 2268b612a6..6421fb6a04 100644
--- a/dynamodb/src/main/java/site/ycsb/db/DynamoDBClient.java
+++ b/dynamodb/src/main/java/site/ycsb/db/DynamoDBClient.java
@@ -16,25 +16,20 @@
 
 package site.ycsb.db;
 
-import com.amazonaws.AmazonClientException;
-import com.amazonaws.AmazonServiceException;
-import com.amazonaws.ClientConfiguration;
-import com.amazonaws.auth.AWSStaticCredentialsProvider;
-import com.amazonaws.auth.PropertiesCredentials;
-import com.amazonaws.client.builder.AwsClientBuilder;
-import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
-import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
-import com.amazonaws.services.dynamodbv2.model.*;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
+import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.http.apache.ApacheHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClientBuilder;
+import software.amazon.awssdk.services.dynamodb.model.*;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import site.ycsb.*;
 
-import java.io.File;
-import java.util.HashMap;
-import java.util.Map;
+import java.net.URI;
+import java.util.*;
 import java.util.Map.Entry;
-import java.util.Set;
-import java.util.Vector;
 
 /**
  * DynamoDB client for YCSB.
@@ -54,7 +49,7 @@ private enum PrimaryKeyType {
     HASH_AND_RANGE
   }
 
-  private AmazonDynamoDB dynamoDB;
+  private volatile DynamoDbClient dynamoDbClient;
   private boolean useLegacyAPI = false;
   private String primaryKeyName;
   private PrimaryKeyType primaryKeyType = PrimaryKeyType.HASH;
@@ -67,14 +62,14 @@ private enum PrimaryKeyType {
   private String ttlKeyName = null;
   private long ttlDuration;
 
+  private boolean useBatchInsert = false;
   private boolean consistentRead = false;
   private boolean inclusiveScan = true;
-  private String region = "us-east-1";
-  private String endpoint = null;
-  private int maxConnects = 50;
   private static final Logger LOGGER = Logger.getLogger(DynamoDBClient.class);
   private static final Status CLIENT_ERROR = new Status("CLIENT_ERROR", "An error occurred on the client.");
   private static final String DEFAULT_HASH_KEY_VALUE = "YCSB_0";
+  private static final int DYNAMODB_BATCH_WRITE_MAX_SIZE = 25;
+  private final Map<String, List<WriteRequest>> batchInsertItemBuffer = new HashMap<>();
 
   @Override
   public void init() throws DBException {
@@ -84,20 +79,17 @@ public void init() throws DBException {
       LOGGER.setLevel(Level.DEBUG);
     }
 
-    String configuredEndpoint = getProperties().getProperty("dynamodb.endpoint", null);
-    String credentialsFile = getProperties().getProperty("dynamodb.awsCredentialsFile", null);
+    String batchInsert = getProperties().getProperty("dynamodb.batchInsert", null);
     String primaryKey = getProperties().getProperty("dynamodb.primaryKey", null);
     String ttlKeyConfiguration = getProperties().getProperty("dynamodb.ttlKey", null);
     String ttlDurationConfiguration = getProperties().getProperty("dynamodb.ttlDuration", null);
     String primaryKeyTypeString = getProperties().getProperty("dynamodb.primaryKeyType", null);
     String consistentReads = getProperties().getProperty("dynamodb.consistentReads", null);
     String inclusiveScans = getProperties().getProperty("dynamodb.inclusiveScan", null);
-    String connectMax = getProperties().getProperty("dynamodb.connectMax", null);
-    String configuredRegion = getProperties().getProperty("dynamodb.region", null);
     String useLegacy = getProperties().getProperty("dynamodb.useLegacyAPI", null);
 
-    if (null != connectMax) {
-      this.maxConnects = Integer.parseInt(connectMax);
+    if (null != batchInsert && "true".equalsIgnoreCase(batchInsert)) {
+      this.useBatchInsert = true;
     }
 
     if (null != consistentReads && "true".equalsIgnoreCase(consistentReads)) {
@@ -112,10 +104,6 @@ public void init() throws DBException {
       this.useLegacyAPI = true;
     }
 
-    if (null != configuredEndpoint) {
-      this.endpoint = configuredEndpoint;
-    }
-
     if (null != ttlKeyConfiguration && null != ttlDurationConfiguration) {
       this.ttlKeyName = ttlKeyConfiguration;
       this.ttlDuration = Integer.parseInt(ttlDurationConfiguration);
@@ -127,6 +115,7 @@ public void init() throws DBException {
     if (null == primaryKey || primaryKey.length() < 1) {
       throw new DBException("Missing primary key attribute name, cannot continue");
     }
+    this.primaryKeyName = primaryKey;
 
     if (null != primaryKeyTypeString) {
       try {
@@ -151,29 +140,34 @@ public void init() throws DBException {
       this.hashKeyValue = getProperties().getProperty("dynamodb.hashKeyValue", DEFAULT_HASH_KEY_VALUE);
     }
 
-    if (null != configuredRegion && configuredRegion.length() > 0) {
-      region = configuredRegion;
-    }
-
-    try {
-      AmazonDynamoDBClientBuilder dynamoDBBuilder = AmazonDynamoDBClientBuilder.standard();
-      dynamoDBBuilder = null == endpoint ?
-          dynamoDBBuilder.withRegion(this.region) :
-          dynamoDBBuilder.withEndpointConfiguration(
-              new AwsClientBuilder.EndpointConfiguration(this.endpoint, this.region)
-          );
-      dynamoDB = dynamoDBBuilder
-          .withClientConfiguration(
-              new ClientConfiguration()
-                  .withTcpKeepAlive(true)
-                  .withMaxConnections(this.maxConnects)
-          )
-          .withCredentials(new AWSStaticCredentialsProvider(new PropertiesCredentials(new File(credentialsFile))))
-          .build();
-      primaryKeyName = primaryKey;
-      LOGGER.info("dynamodb connection created with " + this.endpoint);
-    } catch (Exception e1) {
-      LOGGER.error("DynamoDBClient.init(): Could not initialize DynamoDB client.", e1);
+    // only create one DynamoDB client, used by all YCSB client threads
+    if (dynamoDbClient == null) {
+      synchronized (DynamoDbClient.class) {
+        if (dynamoDbClient == null) {
+          DynamoDbClientBuilder dynamoDbClientBuilder = DynamoDbClient.builder();
+
+          Region region = Region.US_EAST_1;
+          String configuredRegion = getProperties().getProperty("dynamodb.region", null);
+          if (configuredRegion != null) {
+            region = Region.of(configuredRegion);
+          }
+          dynamoDbClientBuilder.region(region);
+
+          String configuredEndpoint = getProperties().getProperty("dynamodb.endpoint", null);
+          if (configuredEndpoint != null) {
+            dynamoDbClientBuilder.endpointOverride(URI.create(configuredEndpoint));
+          }
+
+          // we create the same number of HTTP threads as there are YCSB threads. YCSB default is "1"
+          String configuredThreadCount = getProperties().getProperty(Client.THREAD_COUNT_PROPERTY, "1");
+
+          dynamoDbClientBuilder.httpClient(ApacheHttpClient.builder()
+              .maxConnections(Integer.parseInt(configuredThreadCount))
+              .tcpKeepAlive(true)
+              .build());
+          this.dynamoDbClient = dynamoDbClientBuilder.build();
+        }
+      }
     }
   }
 
@@ -196,31 +190,34 @@ private Map<String, String> aliasFields(Set<String> fields, String prefix) {
 
   private Status getItem(String table, Map<String, AttributeValue> key, Set<String> fields,
                          Map<String, ByteIterator> result, boolean inScan) {
-    GetItemRequest req = new GetItemRequest(table, key);
+    GetItemRequest.Builder getItemBuilder = GetItemRequest.builder()
+        .key(key)
+        .tableName(table);
+
     if (useLegacyAPI) {
-      req.setAttributesToGet(fields);
+      getItemBuilder.attributesToGet(fields);
     } else if (fields != null && !fields.isEmpty()) {
       Map<String, String> aliases = aliasFields(fields, "#");
-      req.setExpressionAttributeNames(aliases);
-      req.setProjectionExpression(String.join(",", aliases.keySet()));
+      getItemBuilder.expressionAttributeNames(aliases);
+      getItemBuilder.projectionExpression(String.join(",", aliases.keySet()));
     }
     if (!inScan) {
-      req.setConsistentRead(consistentRead);
+      getItemBuilder.consistentRead(consistentRead);
     }
-    GetItemResult res;
 
+    GetItemResponse res;
     try {
-      res = dynamoDB.getItem(req);
-    } catch (AmazonServiceException ex) {
+      res = dynamoDbClient.getItem(getItemBuilder.build());
+    } catch (AwsServiceException ex) {
       LOGGER.error(ex);
       return Status.ERROR;
-    } catch (AmazonClientException ex) {
+    } catch (SdkClientException ex) {
       LOGGER.error(ex);
       return CLIENT_ERROR;
     }
 
-    if (null != res.getItem()) {
-      result.putAll(extractResult(res.getItem()));
+    if (res.hasItem()) {
+      result.putAll(extractResult(res.item()));
       if (!inScan && LOGGER.isDebugEnabled()) {
         LOGGER.debug("Result: " + res.toString());
       }
@@ -235,25 +232,28 @@ private Status getItem(String table, Map<String, AttributeValue> key, Set<String
 
   private Status query(String table, String indexName, Map<String, AttributeValue> key, int recordcount,
                        Set<String> fields, Vector<HashMap<String, ByteIterator>> result) {
-    QueryRequest queryRequest = new QueryRequest(table);
-    queryRequest.setLimit(recordcount);
+    QueryRequest.Builder queryRequestBuilder = QueryRequest.builder()
+        .tableName(table)
+        .limit(recordcount);
     if (!indexName.isEmpty()) {
-      queryRequest.setIndexName(indexName);
+      queryRequestBuilder.indexName(indexName);
     }
     
     if (useLegacyAPI) {
-      queryRequest.setAttributesToGet(fields);
+      queryRequestBuilder.attributesToGet(fields);
+      Map<String, Condition> keyConditions = new HashMap<>();
       for (Map.Entry<String, AttributeValue> attr : key.entrySet()) {
-        Condition keycondition = new Condition()
-            .withComparisonOperator(ComparisonOperator.EQ)
-            .withAttributeValueList(attr.getValue());
-        queryRequest.addKeyConditionsEntry(attr.getKey(), keycondition);
+        Condition keycondition = Condition.builder()
+            .comparisonOperator(ComparisonOperator.EQ)
+            .attributeValueList(attr.getValue()).build();
+        keyConditions.put(attr.getKey(), keycondition);
       }
+      queryRequestBuilder.keyConditions(keyConditions);
     } else {
       Map<String, String> attrNames;
       if (fields != null && !fields.isEmpty()) {
         attrNames = aliasFields(fields, "#");
-        queryRequest.setProjectionExpression(String.join(",", attrNames.keySet()));
+        queryRequestBuilder.projectionExpression(String.join(",", attrNames.keySet()));
       } else {
         attrNames = new HashMap<>();
       }
@@ -266,24 +266,24 @@ private Status query(String table, String indexName, Map<String, AttributeValue>
         keyConditionExpression.append(separator).append(nameAlias).append("=").append(valueAlias);
         separator = " AND ";
       }
-      queryRequest.setExpressionAttributeNames(attrNames);
-      queryRequest.setExpressionAttributeValues(attrValues);
-      queryRequest.setKeyConditionExpression(keyConditionExpression.toString());
+      queryRequestBuilder.expressionAttributeNames(attrNames);
+      queryRequestBuilder.expressionAttributeValues(attrValues);
+      queryRequestBuilder.keyConditionExpression(keyConditionExpression.toString());
     }
 
-    QueryResult queryResult;
+    QueryResponse res;
     try {
-      queryResult = dynamoDB.query(queryRequest);
-    } catch (AmazonServiceException ex) {
+      res = dynamoDbClient.query(queryRequestBuilder.build());
+    } catch (AwsServiceException ex) {
       LOGGER.error(ex);
       return Status.ERROR;
-    } catch (AmazonClientException ex) {
+    } catch (SdkClientException ex) {
       LOGGER.error(ex);
       return CLIENT_ERROR;
     }
 
-    if (queryResult.getCount() > 0 && queryResult.getItems() != null) {
-      for (Map<String, AttributeValue> item : queryResult.getItems()) {
+    if (res.count() > 0 && res.items() != null) {
+      for (Map<String, AttributeValue> item : res.items()) {
         result.add(extractResult(item));
       }
     }
@@ -350,41 +350,43 @@ public Status scan(String table, String startkey, int recordcount,
       // startKey is done, rest to go.
     }
 
-    ScanRequest req = new ScanRequest(table);
+    ScanRequest.Builder scanRequestBuilder = ScanRequest.builder()
+        .tableName(table);
     if (tableIndex[1] != null && !tableIndex[1].isEmpty()) {
-      req.setIndexName(tableIndex[1]);
+      scanRequestBuilder.indexName(tableIndex[1]);
     }
     if (useLegacyAPI) {
-      req.setAttributesToGet(fields);
+      scanRequestBuilder.attributesToGet(fields);
     } else if (fields != null && !fields.isEmpty()) {
       Map<String, String> aliases = aliasFields(fields, "#");
-      req.setExpressionAttributeNames(aliases);
-      req.setProjectionExpression(String.join(",", aliases.keySet()));
+      scanRequestBuilder.expressionAttributeNames(aliases);
+      scanRequestBuilder.projectionExpression(String.join(",", aliases.keySet()));
     }
     while (count < recordcount) {
       if (startKey != null) {
-        req.setExclusiveStartKey(startKey);
+        scanRequestBuilder.exclusiveStartKey(startKey);
       }
-      req.setLimit(recordcount - count);
-      ScanResult res;
+      scanRequestBuilder.limit(recordcount - count);
+      ScanResponse res;
       try {
-        res = dynamoDB.scan(req);
-      } catch (AmazonServiceException ex) {
+        res = dynamoDbClient.scan(scanRequestBuilder.build());
+
+      } catch (AwsServiceException ex) {
         LOGGER.error(ex);
         return Status.ERROR;
-      } catch (AmazonClientException ex) {
+      } catch (SdkClientException ex) {
         LOGGER.error(ex);
         return CLIENT_ERROR;
       }
 
-      count += res.getCount();
-      for (Map<String, AttributeValue> items : res.getItems()) {
+      count += res.count();
+      for (Map<String, AttributeValue> items : res.items()) {
         result.add(extractResult(items));
       }
-      startKey = res.getLastEvaluatedKey();
-      if (null == startKey) {
+      if (!res.hasLastEvaluatedKey()) {
         break;
       }
+      startKey = res.lastEvaluatedKey();
     }
 
     return Status.OK;
@@ -397,53 +399,53 @@ public Status update(String table, String key, Map<String, ByteIterator> values)
       LOGGER.debug("updatekey: " + key + " from table: " + table);
     }
 
-    UpdateItemRequest req = new UpdateItemRequest()
-        .withTableName(table)
-        .withKey(createPrimaryKey(key));
+    UpdateItemRequest.Builder updateItemBuilder = UpdateItemRequest.builder()
+        .key(createPrimaryKey(key))
+        .tableName(table);
 
     if (useLegacyAPI) {
       Map<String, AttributeValueUpdate> attributes = new HashMap<>(values.size());
       for (Entry<String, ByteIterator> val : values.entrySet()) {
-        AttributeValue v = new AttributeValue(val.getValue().toString());
-        attributes.put(val.getKey(), new AttributeValueUpdate().withValue(v).withAction("PUT"));
+        AttributeValue v = AttributeValue.fromS(val.getValue().toString());
+        attributes.put(val.getKey(), AttributeValueUpdate.builder().action(AttributeAction.PUT).value(v).build());
       }
 
       if (null != this.ttlKeyName) {
-        AttributeValue v = new AttributeValue().withN(
+        AttributeValue v = AttributeValue.fromN(
             String.valueOf((System.currentTimeMillis() / 1000L) + this.ttlDuration));
-        attributes.put(this.ttlKeyName, new AttributeValueUpdate().withValue(v).withAction("PUT"));
+        attributes.put(this.ttlKeyName, AttributeValueUpdate.builder().action(AttributeAction.PUT).value(v).build());
       }
-      req.setAttributeUpdates(attributes);
+      updateItemBuilder.attributeUpdates(attributes);
     } else {
       Map<String, String> attrNames = new HashMap<>();
       Map<String, AttributeValue> attrValues = new HashMap<>();
       StringBuilder updateExpression = new StringBuilder();
       String separator = "SET ";
       for (Entry<String, ByteIterator> val : values.entrySet()) {
-        AttributeValue v = new AttributeValue(val.getValue().toString());
+        AttributeValue v = AttributeValue.fromS(val.getValue().toString());
         String nameAlias = addAlias("#", val.getKey(), attrNames);
         String valueAlias = addAlias(":", v, attrValues);
         updateExpression.append(separator).append(nameAlias).append("=").append(valueAlias);
         separator = ",";
       }
       if (null != this.ttlKeyName) {
-        AttributeValue v = new AttributeValue().withN(
+        AttributeValue v = AttributeValue.fromN(
             String.valueOf((System.currentTimeMillis() / 1000L) + this.ttlDuration));
         String nameAlias = addAlias("#", this.ttlKeyName, attrNames);
         String valueAlias = addAlias(":", v, attrValues);
         updateExpression.append(separator).append(nameAlias).append("=").append(valueAlias);
       }
-      req.setExpressionAttributeNames(attrNames);
-      req.setExpressionAttributeValues(attrValues);
-      req.setUpdateExpression(updateExpression.toString());
+      updateItemBuilder.expressionAttributeNames(attrNames);
+      updateItemBuilder.expressionAttributeValues(attrValues);
+      updateItemBuilder.updateExpression(updateExpression.toString());
     }
 
     try {
-      dynamoDB.updateItem(req);
-    } catch (AmazonServiceException ex) {
+      dynamoDbClient.updateItem(updateItemBuilder.build());
+    } catch (AwsServiceException ex) {
       LOGGER.error(ex);
       return Status.ERROR;
-    } catch (AmazonClientException ex) {
+    } catch (SdkClientException ex) {
       LOGGER.error(ex);
       return CLIENT_ERROR;
     }
@@ -459,28 +461,90 @@ public Status insert(String table, String key, Map<String, ByteIterator> values)
 
     Map<String, AttributeValue> attributes = createAttributes(values);
     // adding primary key
-    attributes.put(primaryKeyName, new AttributeValue(key));
+    attributes.put(primaryKeyName, AttributeValue.fromS(key));
     if (primaryKeyType == PrimaryKeyType.HASH_AND_RANGE) {
       // If the primary key type is HASH_AND_RANGE, then what has been put
       // into the attributes map above is the range key part of the primary
       // key, we still need to put in the hash key part here.
-      attributes.put(hashKeyName, new AttributeValue(hashKeyValue));
+      attributes.put(hashKeyName, AttributeValue.fromS(hashKeyValue));
     }
 
     if (null != this.ttlKeyName) {
-      attributes.put(this.ttlKeyName, new AttributeValue().withN(
+      attributes.put(this.ttlKeyName, AttributeValue.fromN(
           String.valueOf((System.currentTimeMillis() / 1000L) + this.ttlDuration)));
     }
+    
+    if (useBatchInsert)  {
+      WriteRequest writeRequest = WriteRequest.builder()
+          .putRequest(PutRequest.builder()
+                  .item(attributes)
+                  .build()).build();
+      if (batchInsertItemBuffer.containsKey(table)) {
+        batchInsertItemBuffer.get(table).add(writeRequest);
+      } else {
+        batchInsertItemBuffer.put(table, new ArrayList<>(Arrays.asList(writeRequest)));
+      }
+      if (batchInsertItemBuffer.values().stream().mapToInt(List::size).sum() == DYNAMODB_BATCH_WRITE_MAX_SIZE) {
+        batchInsertItems();
+      }
+    } else {
+      PutItemRequest putItemRequest = PutItemRequest.builder().item(attributes).tableName(table).build();
+      try {
+        dynamoDbClient.putItem(putItemRequest);
+      } catch (AwsServiceException ex) {
+        LOGGER.error(ex);
+        return Status.ERROR;
+      } catch (SdkClientException ex) {
+        LOGGER.error(ex);
+        return CLIENT_ERROR;
+      }
+    }
+    return Status.OK;
+  }
 
-    PutItemRequest putItemRequest = new PutItemRequest(table, attributes);
-    try {
-      dynamoDB.putItem(putItemRequest);
-    } catch (AmazonServiceException ex) {
-      LOGGER.error(ex);
-      return Status.ERROR;
-    } catch (AmazonClientException ex) {
-      LOGGER.error(ex);
-      return CLIENT_ERROR;
+  private Status batchInsertItems() {
+    if (LOGGER.isDebugEnabled()) {
+      LOGGER.debug("batch write of " + batchInsertItemBuffer.values().stream().mapToInt(List::size).sum() + " items");
+    }
+    if (batchInsertItemBuffer.values().stream().mapToInt(List::size).sum() > 0) {
+      try {
+        boolean allItemsInserted = false;
+        int writeAttempts = 0;
+
+        // try to put the items in the DynamoDB table
+        // retry with an exponential backoff if the DynamoDB client
+        // could not process all items successfully
+        while (!allItemsInserted && writeAttempts <= 5) {
+          BatchWriteItemResponse response = dynamoDbClient.batchWriteItem(BatchWriteItemRequest.builder()
+              .requestItems(batchInsertItemBuffer)
+              .build());
+          writeAttempts++;
+          batchInsertItemBuffer.clear();
+          if (response.hasUnprocessedItems() &&
+              response.unprocessedItems().values().stream().mapToInt(List::size).sum() > 0) {
+            batchInsertItemBuffer.putAll(response.unprocessedItems());
+            try {
+              Thread.sleep(250 * (int)Math.pow(2, writeAttempts));
+            } catch (InterruptedException ie) {
+              if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug("batchWriteItem sleep for exponential backoff & retry got interrupted");
+              }
+            }
+          } else {
+            allItemsInserted = true;
+          }
+        }
+        if (!allItemsInserted) {
+          LOGGER.error("Client failed to insert all the items in the batch");
+          return CLIENT_ERROR;
+        }
+      } catch (AwsServiceException ex) {
+        LOGGER.error(ex);
+        return Status.ERROR;
+      } catch (SdkClientException ex) {
+        LOGGER.error(ex);
+        return CLIENT_ERROR;
+      }
     }
     return Status.OK;
   }
@@ -492,24 +556,36 @@ public Status delete(String table, String key) {
       LOGGER.debug("deletekey: " + key + " from table: " + table);
     }
 
-    DeleteItemRequest req = new DeleteItemRequest(table, createPrimaryKey(key));
+    DeleteItemRequest req = DeleteItemRequest.builder().key(createPrimaryKey(key)).tableName(table).build();
 
     try {
-      dynamoDB.deleteItem(req);
-    } catch (AmazonServiceException ex) {
+      dynamoDbClient.deleteItem(req);
+    } catch (AwsServiceException ex) {
       LOGGER.error(ex);
       return Status.ERROR;
-    } catch (AmazonClientException ex) {
+    } catch (SdkClientException ex) {
       LOGGER.error(ex);
       return CLIENT_ERROR;
     }
     return Status.OK;
   }
 
+  /**
+   * Cleanup any state for this DB.
+   * Called once per DB instance; there is one DB instance per client thread.
+   */
+  @Override
+  public void cleanup() throws DBException {
+    if (useBatchInsert) {
+      // flush any remaining items to the DynamoDB table
+      batchInsertItems();
+    }
+  }
+
   private static Map<String, AttributeValue> createAttributes(Map<String, ByteIterator> values) {
     Map<String, AttributeValue> attributes = new HashMap<>(values.size() + 1);
     for (Entry<String, ByteIterator> val : values.entrySet()) {
-      attributes.put(val.getKey(), new AttributeValue(val.getValue().toString()));
+      attributes.put(val.getKey(), AttributeValue.fromS(val.getValue().toString()));
     }
     return attributes;
   }
@@ -524,7 +600,7 @@ private HashMap<String, ByteIterator> extractResult(Map<String, AttributeValue>
       if (LOGGER.isDebugEnabled()) {
         LOGGER.debug(String.format("Result- key: %s, value: %s", attr.getKey(), attr.getValue()));
       }
-      rItems.put(attr.getKey(), new StringByteIterator(attr.getValue().getS()));
+      rItems.put(attr.getKey(), new StringByteIterator(attr.getValue().s()));
     }
     return rItems;
   }
@@ -532,10 +608,10 @@ private HashMap<String, ByteIterator> extractResult(Map<String, AttributeValue>
   private Map<String, AttributeValue> createPrimaryKey(String key) {
     Map<String, AttributeValue> k = new HashMap<>();
     if (primaryKeyType == PrimaryKeyType.HASH) {
-      k.put(primaryKeyName, new AttributeValue().withS(key));
+      k.put(primaryKeyName, AttributeValue.fromS(key));
     } else if (primaryKeyType == PrimaryKeyType.HASH_AND_RANGE) {
-      k.put(hashKeyName, new AttributeValue().withS(hashKeyValue));
-      k.put(primaryKeyName, new AttributeValue().withS(key));
+      k.put(hashKeyName, AttributeValue.fromS(hashKeyValue));
+      k.put(primaryKeyName, AttributeValue.fromS(key));
     } else {
       throw new RuntimeException("Assertion Error: impossible primary key type");
     }
diff --git a/dynamodb/src/test/java/site/ycsb/db/dynamodb/DynamoDBClientTest.java b/dynamodb/src/test/java/site/ycsb/db/dynamodb/DynamoDBClientTest.java
index ca4c3245e0..98411511a4 100644
--- a/dynamodb/src/test/java/site/ycsb/db/dynamodb/DynamoDBClientTest.java
+++ b/dynamodb/src/test/java/site/ycsb/db/dynamodb/DynamoDBClientTest.java
@@ -21,15 +21,14 @@
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
-
-import com.amazonaws.auth.AWSStaticCredentialsProvider;
-import com.amazonaws.auth.BasicAWSCredentials;
-import com.amazonaws.client.builder.AwsClientBuilder;
-import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
-import com.amazonaws.services.dynamodbv2.AmazonDynamoDBClientBuilder;
-import com.amazonaws.services.dynamodbv2.model.*;
+
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.model.*;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.core.SdkSystemSetting;
+
 import com.google.common.collect.Sets;
 
 import org.junit.*;
@@ -49,6 +48,8 @@
 import java.util.Set;
 import java.util.Vector;
 import java.util.stream.Collectors;
+import java.net.URI;
+
 
 /**
  * Integration tests for the DynamoDB client using Scylla's Alternator interface
@@ -56,13 +57,15 @@
 public class DynamoDBClientTest {
   private static String HOST;
   private static int PORT;
-  private static java.io.File tempCredFile;
 
   public static ScyllaDBContainer scyllaContainer;
 
   @BeforeClass
   public static void setUpContainer() {
     try {
+      // Set dummy AWS credentials for testing
+      System.setProperty(SdkSystemSetting.AWS_ACCESS_KEY_ID.property(), "dummy");
+      System.setProperty(SdkSystemSetting.AWS_SECRET_ACCESS_KEY.property(), "dummy");
       // Determine Scylla image from system properties, with sensible defaults
       // -Dscylla.image takes precedence. Otherwise built from -Dscylla.version (defaults to 2025.1)
       final String version = System.getProperty("scylla.version", "2025.1");
@@ -82,18 +85,6 @@ public static void setUpContainer() {
     } catch (Throwable t) {
       Assume.assumeTrue("Skipping DynamoDB tests because Docker/Testcontainers is not available: " + t.getMessage(), false);
     }
-
-    // Create a temporary credentials file for the test
-    try {
-      tempCredFile = java.io.File.createTempFile("aws-credentials", ".properties");
-      tempCredFile.deleteOnExit();
-      try (java.io.FileWriter writer = new java.io.FileWriter(tempCredFile)) {
-        writer.write("accessKey=dummy\n");
-        writer.write("secretKey=dummy\n");
-      }
-    } catch (java.io.IOException e) {
-      throw new RuntimeException("Failed to create temporary credentials file", e);
-    }
   }
 
   @AfterClass
@@ -102,10 +93,13 @@ public static void tearDownContainer() {
       scyllaContainer.stop();
       scyllaContainer = null;
     }
+    // Clean up system properties
+    System.clearProperty(SdkSystemSetting.AWS_ACCESS_KEY_ID.property());
+    System.clearProperty(SdkSystemSetting.AWS_SECRET_ACCESS_KEY.property());
   }
 
   private DynamoDBClient ycsbClient;
-  private AmazonDynamoDB awsClient;
+  private DynamoDbClient awsClient;
 
   @Rule
   public TestName testName = new TestName();
@@ -116,37 +110,41 @@ private String TABLE() {
   private void createTable() throws InterruptedException {
     try {
       // Create DynamoDB table
-      CreateTableRequest createTableRequest = new CreateTableRequest()
-          .withTableName(TABLE())
-          .withKeySchema(new KeySchemaElement("y_id", KeyType.HASH))
-          .withBillingMode(BillingMode.PAY_PER_REQUEST);
+      CreateTableRequest.Builder createTableRequestBuilder = CreateTableRequest.builder()
+          .tableName(TABLE())
+          .keySchema(KeySchemaElement.builder()
+              .attributeName("y_id")
+              .keyType(KeyType.HASH)
+              .build())
+          .billingMode(BillingMode.PAY_PER_REQUEST);
       if (testName.getMethodName().contains("SecondaryIndex")) {
-        createTableRequest = createTableRequest.withAttributeDefinitions(
-              new AttributeDefinition("y_id", ScalarAttributeType.S),
-              new AttributeDefinition(fieldName(0), ScalarAttributeType.S)
-            ).withGlobalSecondaryIndexes(new GlobalSecondaryIndex()
-              .withIndexName("field0-index")
-              .withKeySchema(new KeySchemaElement(fieldName(0), KeyType.HASH))
-              .withProjection(new Projection().withProjectionType(ProjectionType.ALL)));
+        createTableRequestBuilder.attributeDefinitions(
+              AttributeDefinition.builder().attributeName("y_id").attributeType(ScalarAttributeType.S).build(),
+              AttributeDefinition.builder().attributeName(fieldName(0)).attributeType(ScalarAttributeType.S).build()
+            ).globalSecondaryIndexes(GlobalSecondaryIndex.builder()
+              .indexName("field0-index")
+              .keySchema(KeySchemaElement.builder().attributeName(fieldName(0)).keyType(KeyType.HASH).build())
+              .projection(Projection.builder().projectionType(ProjectionType.ALL).build())
+              .build());
       } else {
-        createTableRequest = createTableRequest.withAttributeDefinitions(
-              new AttributeDefinition("y_id", ScalarAttributeType.S));
+        createTableRequestBuilder.attributeDefinitions(
+              AttributeDefinition.builder().attributeName("y_id").attributeType(ScalarAttributeType.S).build());
       }
-      awsClient.createTable(createTableRequest);
+      awsClient.createTable(createTableRequestBuilder.build());
     } catch (ResourceInUseException e) {
       // Table already exists, ignore
     }
-      
+  
     // Wait for table to be active - simple polling approach
     int attempts = 0;
     while (attempts < 30) {
       try {
-        TableDescription result = awsClient.describeTable(TABLE()).getTable();
-        boolean tableActive = "ACTIVE".equals(result.getTableStatus());
-        boolean indexActive = result.getGlobalSecondaryIndexes() == null
-          || result.getGlobalSecondaryIndexes().stream().allMatch(
-                      gsi -> "ACTIVE".equals(gsi.getIndexStatus()));
-        
+        TableDescription result = awsClient.describeTable(
+          DescribeTableRequest.builder().tableName(TABLE()).build()).table();
+        boolean tableActive = TableStatus.ACTIVE.equals(result.tableStatus());
+        boolean indexActive = result.globalSecondaryIndexes() == null
+          || result.globalSecondaryIndexes().stream().allMatch(
+                      gsi -> TableStatus.ACTIVE.equals(gsi.indexStatus()));
         if (tableActive && indexActive) {
           return;
         }
@@ -159,11 +157,11 @@ private void createTable() throws InterruptedException {
   @Before
   public void setUpTable() throws Exception {
     // Create DynamoDB client pointing to Scylla's Alternator interface
-    BasicAWSCredentials awsCreds = new BasicAWSCredentials("dummy", "dummy");
-    awsClient = AmazonDynamoDBClientBuilder.standard()
-        .withCredentials(new AWSStaticCredentialsProvider(awsCreds))
-        .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(
-            "http://" + HOST + ":" + PORT, "us-east-1"))
+    awsClient = DynamoDbClient.builder()
+        .credentialsProvider(StaticCredentialsProvider.create(
+          AwsBasicCredentials.builder().accessKeyId("dummy").secretAccessKey("dummy").build()))
+        .endpointOverride(URI.create("http://" + HOST + ":" + PORT))
+        .region(Region.of("us-east-1"))
         .build();
 
     // Create table
@@ -173,7 +171,7 @@ public void setUpTable() throws Exception {
   @After
   public void clearTable() {
     try {
-      awsClient.deleteTable(TABLE());
+      awsClient.deleteTable(DeleteTableRequest.builder().tableName(TABLE()).build());
     } catch (Exception e) {
       // Ignore cleanup errors
     }
@@ -183,7 +181,6 @@ public void clearTable() {
   public void setUpClient() throws Exception {
     Properties p = new Properties();
     p.setProperty("dynamodb.endpoint", "http://" + HOST + ":" + PORT);
-    p.setProperty("dynamodb.awsCredentialsFile", tempCredFile.getAbsolutePath());
     p.setProperty("dynamodb.primaryKey", "y_id");
     p.setProperty("dynamodb.region", "us-east-1");
     p.setProperty("table", TABLE());
@@ -222,13 +219,14 @@ private String keyValue(int rowNum) {
   private void insertRows(int count) {
     for (int i = 0; i < count; i++) {
       Map<String, AttributeValue> item = new HashMap<>();
-      item.put("y_id", new AttributeValue(keyValue(i)));
-      item.put(fieldName(0), new AttributeValue(fieldValue(i, 0)));
-      item.put(fieldName(1), new AttributeValue(fieldValue(i, 1)));
-
-      PutItemRequest putItemRequest = new PutItemRequest()
-          .withTableName(TABLE())
-          .withItem(item);
+      item.put("y_id", AttributeValue.fromS(keyValue(i)));
+      item.put(fieldName(0), AttributeValue.fromS(fieldValue(i, 0)));
+      item.put(fieldName(1), AttributeValue.fromS(fieldValue(i, 1)));
+
+      PutItemRequest putItemRequest = PutItemRequest.builder()
+          .tableName(TABLE())
+          .item(item)
+          .build();
       awsClient.putItem(putItemRequest);
     }
   }
@@ -289,15 +287,28 @@ public void testInsert() {
     final Status status = ycsbClient.insert(TABLE(), key, StringByteIterator.getByteIteratorMap(input));
     assertThat(status, is(Status.OK));
 
+    if (ycsbClient.getProperties().getProperty("dynamodb.batchInsert", "false").equals("true")) {
+      try {
+        ycsbClient.cleanup(); // Ensure batch is flushed
+      } catch (Exception e) { }
+    }
+
     // Verify result
-    GetItemRequest getItemRequest = new GetItemRequest()
-        .withTableName(TABLE())
-        .withKey(Map.of("y_id", new AttributeValue(key)));
-
-    GetItemResult getItemResult = awsClient.getItem(getItemRequest);
-    assertThat(getItemResult.getItem(), notNullValue());
-    assertThat(getItemResult.getItem().get(fieldName(0)).getS(), is(fieldValue(0)));
-    assertThat(getItemResult.getItem().get(fieldName(1)).getS(), is(fieldValue(1)));
+    GetItemRequest getItemRequest = GetItemRequest.builder()
+        .tableName(TABLE())
+        .key(Map.of("y_id", AttributeValue.fromS(key)))
+        .build();
+
+    GetItemResponse getItemResult = awsClient.getItem(getItemRequest);
+    assertThat(getItemResult.hasItem(), is(true));
+    assertThat(getItemResult.item().get(fieldName(0)).s(), is(fieldValue(0)));
+    assertThat(getItemResult.item().get(fieldName(1)).s(), is(fieldValue(1)));
+  }
+
+  @Test
+  public void testBatchInsert() {
+    updateClientProperties(Map.of("dynamodb.batchInsert", "true"));
+    testInsert();
   }
 
   @Test
@@ -312,14 +323,15 @@ public void testUpdate() {
     assertThat(status, is(Status.OK));
 
     // Verify result
-    GetItemRequest getItemRequest = new GetItemRequest()
-        .withTableName(TABLE())
-        .withKey(Map.of("y_id", new AttributeValue(keyValue())));
-
-    GetItemResult getItemResult = awsClient.getItem(getItemRequest);
-    assertThat(getItemResult.getItem(), notNullValue());
-    assertThat(getItemResult.getItem().get(fieldName(0)).getS(), is("new-value1"));
-    assertThat(getItemResult.getItem().get(fieldName(1)).getS(), is("new-value2"));
+    GetItemRequest getItemRequest = GetItemRequest.builder()
+        .tableName(TABLE())
+        .key(Map.of("y_id", AttributeValue.fromS(keyValue())))
+        .build();
+
+    GetItemResponse getItemResult = awsClient.getItem(getItemRequest);
+    assertThat(getItemResult.hasItem(), is(true));
+    assertThat(getItemResult.item().get(fieldName(0)).s(), is("new-value1"));
+    assertThat(getItemResult.item().get(fieldName(1)).s(), is("new-value2"));
   }
 
   @Test
@@ -330,12 +342,13 @@ public void testDelete() {
     assertThat(status, is(Status.OK));
 
     // Verify result
-    GetItemRequest getItemRequest = new GetItemRequest()
-        .withTableName(TABLE())
-        .withKey(Map.of("y_id", new AttributeValue(keyValue())));
+    GetItemRequest getItemRequest = GetItemRequest.builder()
+        .tableName(TABLE())
+        .key(Map.of("y_id", AttributeValue.fromS(keyValue())))
+        .build();
 
-    GetItemResult getItemResult = awsClient.getItem(getItemRequest);
-    assertThat(getItemResult.getItem(), nullValue());
+    GetItemResponse getItemResult = awsClient.getItem(getItemRequest);
+    assertThat(getItemResult.hasItem(), is(false));
   }
 
   private void assertRows(int expected, Vector<HashMap<String, ByteIterator>> results) {