如何将AWS S3中的CSV导入DynamoDB？及导入失败问题排查

一、当前导入失败的错误分析 #

你遇到的Value for Item.Score is ambiguous错误，核心原因是CSV中Score列的数据类型不统一，DynamoDB无法确定该字段的类型（比如部分行是数字、部分是字符串，或存在空值、格式异常内容）。

排查步骤：

• 打开CSVs/atl22.csv，检查表头下第一行（itemIndex=0）的Score值，确认是否存在多余引号、混合数字文本、空值写法不一致等格式问题
• 遍历整个CSV，确保Score列数据类型统一：要么全为数字（不加引号），要么全为字符串（若为字符串类型，导入时需明确指定）
• 同步检查分区键匹配问题：确认DynamoDB表atl_snaps的分区键（含排序键若有）名称，与CSV表头对应列名完全一致（大小写敏感），且对应列无空值

二、程序化从S3导入CSV到DynamoDB的方法 #

以下是两种实用方案：

方案1：调用DynamoDB批量导入API（推荐，适合大量数据） #

直接通过boto3发起导入任务，与控制台操作逻辑一致：

import boto3

def start_dynamodb_import(table_name, s3_bucket, s3_prefix, region='us-west-2'):
    dynamodb = boto3.client('dynamodb', region_name=region)
    response = dynamodb.import_table(
        S3BucketSource={
            'S3Bucket': s3_bucket,
            'S3Prefix': s3_prefix
        },
        TableCreationParameters={
            'TableName': table_name,
            'AttributeDefinitions': [
                # 根据表结构定义字段，示例：分区键为TeamID（字符串类型）
                {'AttributeName': 'TeamID', 'AttributeType': 'S'},
                {'AttributeName': 'Score', 'AttributeType': 'N'} # 明确Score为数字类型
            ],
            'KeySchema': [
                {'AttributeName': 'TeamID', 'KeyType': 'HASH'} # 分区键
                # 若有排序键，添加：{'AttributeName': 'Year', 'KeyType': 'RANGE'}
            ],
            'ProvisionedThroughput': {'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
        },
        InputFormat='CSV',
        InputFormatOptions={
            'Csv': {
                'HeaderList': ['TeamID', 'Score', 'PlayerName'] # 指定CSV表头，与文件列对应
            }
        }
    )
    print(f"导入任务已启动，ARN: {response['ImportTableDescription']['ImportArn']}")
    return response

# 调用示例：导入atl22.csv到atl_snaps表
start_dynamodb_import(
    table_name='atl_snaps',
    s3_bucket='teamcsvs',
    s3_prefix='CSVs/atl22.csv',
    region='us-west-2'
)

方案2：读取S3 CSV内容后批量写入（适合自定义清洗逻辑） #

若需先清洗数据再导入，可读取S3文件内容后用批量写入接口提交：

import boto3
import csv
from io import StringIO

def s3_csv_to_dynamodb(table_name, s3_bucket, s3_key, region='us-west-2'):
    s3 = boto3.client('s3', region_name=region)
    dynamodb = boto3.resource('dynamodb', region_name=region)
    table = dynamodb.Table(table_name)
    
    # 读取S3中的CSV文件
    response = s3.get_object(Bucket=s3_bucket, Key=s3_key)
    csv_content = response['Body'].read().decode('utf-8')
    csv_reader = csv.DictReader(StringIO(csv_content))
    
    # 每25条提交一次（DynamoDB批量写入上限）
    items = []
    for row in csv_reader:
        # 清洗数据示例：将Score转为数字，空值设为0
        row['Score'] = int(row['Score'].strip()) if row['Score'].strip() else 0
        items.append({'PutRequest': {'Item': row}})
        
        if len(items) == 25:
            table.batch_writer().batch_write_item(RequestItems={table_name: items})
            items = []
    # 处理剩余数据
    if items:
        table.batch_writer().batch_write_item(RequestItems={table_name: items})
    print(f"文件{s3_key}数据导入完成")

# 调用示例
s3_csv_to_dynamodb(
    table_name='atl_snaps',
    s3_bucket='teamcsvs',
    s3_key='CSVs/atl22.csv',
    region='us-west-2'
)

三、额外注意事项 #

• CSV文件建议用UTF-8编码，避免乱码引发导入错误
• 若CSV含空值，需确认DynamoDB表是否允许该字段为空，或统一设置默认值
• 批量导入时需注意DynamoDB读写容量限制，避免触发限流

内容的提问来源于stack exchange，提问作者8bitarson