如何在Android Kotlin中用BatchWriteItem或批量自定义解析器上传数据到AWS DynamoDB

云原生阿岚

2026-6-26

Android Kotlin + AWS Amplify 批量同步数据到DynamoDB解决方案

问题背景

我有一个基于Android Kotlin的项目，用到了AWS Amplify的Cognito、Datastore、Appsync等服务。作为AWS新手，我被批量同步数据到DynamoDB的问题困扰了一个月，目标是用最小延迟同步约2000条记录。

已尝试两种逐条同步的方法，均存在高延迟问题：

用Amplify DataStore逐条保存：2000条耗时约20分钟
用Amplify API.mutate逐条突变：2000条耗时约200秒

核心问题是逐条请求导致的网络 overhead 过高，需要实现批量上传方案。

可行解决方案

方案一：自定义GraphQL批量突变（推荐，贴合Amplify生态）

通过在AppSync后端定义批量操作接口，生成客户端代码后调用，性能远高于逐条请求。

步骤1：修改GraphQL Schema

在项目的schema.graphql中添加批量创建字段：

type Mutation {
  createRandomNumber(input: CreateRandomNumberInput!): RandomNumber
  # 添加批量创建接口
  batchCreateRandomNumbers(inputs: [CreateRandomNumberInput!]!): [RandomNumber]
}

步骤2：配置AppSync批量解析器

在AWS控制台进入你的AppSync API，为batchCreateRandomNumbers配置DynamoDB批量写入解析器：

数据源选择你的DynamoDB表
请求映射模板（将输入转为BatchWriteItem格式）：

{
  "version": "2017-02-28",
  "operation": "BatchWriteItem",
  "tables": {
    "RandomNumbersTable": [
      #foreach($item in $ctx.args.inputs)
        {
          "PutRequest": {
            "Item": {
              "id": { "S": "$util.autoId()" },
              "number": { "N": "$item.number" }
              # 其他字段按需添加
            }
          }
        }#if($foreach.hasNext),#end
      #end
    ]
  }
}

响应映射模板：

$util.toJson($ctx.result.data.RandomNumbersTable)

步骤3：生成Android客户端代码

在项目根目录执行命令，生成批量操作对应的Kotlin代码：

amplify codegen models

步骤4：Android端调用批量接口

按DynamoDB限制（单次最多25条）拆分数据，批量调用：

// 生成2000条数据
val items = (1..2000).map {
    CreateRandomNumberInput.builder()
        .number(it)
        .build()
}

// 按25条为一批拆分并发送请求
items.chunked(25).forEach { batch ->
    Amplify.API.mutate(
        BatchCreateRandomNumbersRequest.builder()
            .inputs(batch)
            .build(),
        { response ->
            Log.i("BatchSync", "成功同步批次：${response.data?.size}条")
        },
        { error ->
            Log.e("BatchSync", "批次同步失败", error)
        }
    )
}

该方案2000条数据仅需80次请求，通常几十秒内即可完成。

方案二：直接使用DynamoDB SDK for Android

如果不想修改GraphQL Schema，可直接调用DynamoDB原生BatchWriteItem接口，复用Amplify的身份凭证。

步骤1：添加依赖

在app/build.gradle中添加DynamoDB SDK依赖：

dependencies {
    implementation 'com.amazonaws:aws-android-sdk-dynamodb:2.62.0'
}

步骤2：初始化DynamoDB客户端

复用Amplify的身份凭证，初始化客户端：

val credentialsProvider = Amplify.Auth.getCredentialsProvider()
val dynamoDbClient = AmazonDynamoDBClientBuilder.standard()
    .withCredentials(credentialsProvider)
    .withRegion(Regions.US_EAST_1) // 替换为你的AWS区域
    .build()

步骤3：执行批量写入

按25条为一批拆分数据，发送批量请求：

val items = (1..2000).map { number ->
    val item = mutableMapOf<String, AttributeValue>()
    item["id"] = AttributeValue().withS(UUID.randomUUID().toString())
    item["number"] = AttributeValue().withN(number.toString())
    PutRequest().withItem(item)
}

items.chunked(25).forEach { batch ->
    val requestItems = mutableMapOf<String, List<WriteRequest>>()
    requestItems["RandomNumbersTable"] = batch.map { WriteRequest().withPutRequest(it) }
    
    val batchRequest = BatchWriteItemRequest().withRequestItems(requestItems)
    dynamoDbClient.batchWriteItem(batchRequest)
    
    Log.i("BatchSync", "批次处理完成")
}

注意：需确保Cognito身份池权限包含dynamodb:BatchWriteItem。

方案三：优化现有API并发请求（临时应急）

如果无法修改后端，可通过控制并发数提升逐条请求的速度，避免被限流：

val items = (1..2000).map {
    RandomNumbers.builder()
        .number(it)
        .build()
}

// 用协程控制并发数（比如同时执行10条请求）
val semaphore = Semaphore(10)
CoroutineScope(Dispatchers.IO).launch {
    items.forEach { item ->
        semaphore.acquire()
        launch {
            try {
                Amplify.API.mutate(
                    ModelMutation.create(item),
                    { Log.i("BatchSync", "已保存：${it.data?.number}") },
                    { Log.e("BatchSync", "保存失败", it) }
                )
            } finally {
                semaphore.release()
            }
        }
    }
}

该方案性能不如前两种，但比纯逐条同步快很多。

内容的提问来源于stack exchange，提问作者Mahesh Gouda