如何用Terraform和Docker管控本地及云端DynamoDB并复用表定义

核心思路：单一数据源复用 #

将DynamoDB表的所有核心定义（表名、Hash Key、Range Key、属性、GSI等）集中存储在JSON/YAML格式的配置文件中，让Terraform和本地应用代码都读取这个文件来创建表。这样既避免了重复维护多份定义，也能保证本地测试环境与云端的表结构完全一致。

1. 编写统一的表定义配置文件 #

创建一个dynamodb-table-configs.yaml（或JSON），把所有表的结构定义集中在这里：

tables:
  - name: user-profiles
    hash_key:
      name: user_id
      type: S
    range_key:
      name: created_at
      type: N
    attribute_definitions:
      - name: user_id
        type: S
      - name: created_at
        type: N
      - name: email
        type: S
    global_secondary_indexes:
      - name: email-index
        hash_key: email
        projection_type: ALL
        read_capacity: 5
        write_capacity: 5

  - name: order-history
    hash_key:
      name: order_id
      type: S
    attribute_definitions:
      - name: order_id
        type: S
      - name: user_id
        type: S
    global_secondary_indexes:
      - name: user-order-index
        hash_key: user_id
        projection_type: KEYS_ONLY
        read_capacity: 3
        write_capacity: 3

2. Terraform读取配置文件创建云端表 #

在Terraform代码中，通过yamldecode()（或jsondecode()）读取配置文件，用for_each批量创建表：

locals {
  dynamodb_table_configs = yamldecode(file("./dynamodb-table-configs.yaml")).tables
}

resource "aws_dynamodb_table" "app_tables" {
  for_each = { for table in local.dynamodb_table_configs : table.name => table }

  name           = each.value.name
  billing_mode   = "PAY_PER_REQUEST" # 按需付费，可根据业务调整为PROVISIONED
  hash_key       = each.value.hash_key.name
  range_key      = lookup(each.value, "range_key", null) != null ? each.value.range_key.name : null

  # 批量定义属性
  dynamic "attribute" {
    for_each = each.value.attribute_definitions
    content {
      name = attribute.value.name
      type = attribute.value.type
    }
  }

  # 批量创建GSI
  dynamic "global_secondary_index" {
    for_each = lookup(each.value, "global_secondary_indexes", [])
    content {
      name               = global_secondary_index.value.name
      hash_key           = global_secondary_index.value.hash_key
      projection_type    = global_secondary_index.value.projection_type
      read_capacity      = lookup(global_secondary_index.value, "read_capacity", null)
      write_capacity     = lookup(global_secondary_index.value, "write_capacity", null)
    }
  }
}

3. 本地应用读取配置文件创建测试表 #

在本地应用的启动脚本（以Python为例）中，读取同一个配置文件，调用DynamoDB本地镜像的API创建表：

import boto3
import yaml

def init_local_dynamodb():
    # 连接本地Docker运行的DynamoDB
    dynamodb = boto3.resource('dynamodb', endpoint_url='http://localhost:8000')
    
    # 读取统一配置文件
    with open('./dynamodb-table-configs.yaml', 'r') as f:
        config = yaml.safe_load(f)

    for table_def in config['tables']:
        # 构建Key Schema
        key_schema = [{'AttributeName': table_def['hash_key']['name'], 'KeyType': 'HASH'}]
        if 'range_key' in table_def:
            key_schema.append({'AttributeName': table_def['range_key']['name'], 'KeyType': 'RANGE'})
        
        # 构建属性定义
        attr_defs = [
            {'AttributeName': attr['name'], 'AttributeType': attr['type']}
            for attr in table_def['attribute_definitions']
        ]
        
        # 构建GSI配置
        gsi_configs = []
        if 'global_secondary_indexes' in table_def:
            for gsi in table_def['global_secondary_indexes']:
                gsi_schema = [{'AttributeName': gsi['hash_key'], 'KeyType': 'HASH'}]
                gsi_configs.append({
                    'IndexName': gsi['name'],
                    'KeySchema': gsi_schema,
                    'Projection': {'ProjectionType': gsi['projection_type']},
                    'ProvisionedThroughput': {
                        'ReadCapacityUnits': gsi.get('read_capacity', 5),
                        'WriteCapacityUnits': gsi.get('write_capacity', 5)
                    }
                })
        
        # 创建本地表
        table = dynamodb.create_table(
            TableName=table_def['name'],
            KeySchema=key_schema,
            AttributeDefinitions=attr_defs,
            GlobalSecondaryIndexes=gsi_configs,
            ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
        )
        table.wait_until_exists()
        print(f"本地表 {table_def['name']} 创建完成")

if __name__ == '__main__':
    init_local_dynamodb()

额外优化建议 #

• 如果需要区分本地与云端的配置差异（比如读写容量），可以在配置文件中添加local和cloud的分支，读取时根据环境变量切换。
• 可以把配置文件放在项目根目录的config文件夹下，方便团队成员统一获取。

内容的提问来源于stack exchange，提问作者Roger Thomas