如何将MWAA与DynamoDB集成？替换S3桶提升MWAA性能

1. 核心集成场景与思路 #

替换S3的核心目标是利用DynamoDB的低延迟、高读写吞吐量特性，优化以下MWAA场景的扩展性与性能：

• 结构化任务状态/结果存储
• DAG配置参数存储
• 任务间小体量数据缓存

这些场景下，DynamoDB比S3更适合高频读写、快速查询的需求。

2. 必备权限配置 #

为MWAA的执行角色添加DynamoDB读写权限，在IAM策略中新增如下规则：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "dynamodb:GetItem",
                "dynamodb:PutItem",
                "dynamodb:UpdateItem",
                "dynamodb:DeleteItem",
                "dynamodb:Query",
                "dynamodb:Scan"
            ],
            "Resource": "arn:aws:dynamodb:你的AWS区域:你的账号ID:table/目标表名"
        }
    ]
}

将该策略附加到MWAA执行角色，确保Airflow任务能正常访问DynamoDB。

3. Airflow任务集成实现 #

方法1：使用boto3直接操作 #

在PythonOperator中编写代码，直接替换S3的上传/下载逻辑：

from airflow import DAG
from airflow.operators.python import PythonOperator
import boto3
from datetime import datetime

def write_task_result_to_ddb():
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.Table('mwaa_task_results')
    # 写入任务结果
    table.put_item(
        Item={
            'task_id': 'sample_data_processing',
            'execution_date': str(datetime.now()),
            'status': 'completed',
            'output_stats': {'processed_rows': 1000, 'error_count': 0}
        }
    )

def query_task_result_from_ddb():
    dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
    table = dynamodb.Table('mwaa_task_results')
    # 查询指定任务结果
    response = table.get_item(
        Key={
            'task_id': 'sample_data_processing',
            'execution_date': '2024-05-20 14:30:00'
        }
    )
    print("任务结果:", response['Item'])

with DAG('mwaa_ddb_integration', start_date=datetime(2024,5,20), schedule_interval='@hourly') as dag:
    write_task = PythonOperator(
        task_id='write_to_ddb',
        python_callable=write_task_result_to_ddb
    )
    read_task = PythonOperator(
        task_id='read_from_ddb',
        python_callable=query_task_result_from_ddb
    )
    write_task >> read_task

方法2：使用Airflow内置DynamoDBHook（推荐） #

利用Airflow官方提供的DynamoDBHook简化操作，无需手动初始化boto3客户端：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.dynamodb import DynamoDBHook
from datetime import datetime
from boto3.dynamodb.conditions import Key

def use_ddb_hook_for_config():
    hook = DynamoDBHook(aws_conn_id='aws_default')
    table = hook.get_resource_type('dynamodb').Table('mwaa_dag_configs')
    # 写入DAG配置
    table.put_item(
        Item={
            'dag_id': 'mwaa_ddb_integration',
            'config_key': 'data_source',
            'config_value': 's3://source-bucket/data/',
            'updated_at': str(datetime.now())
        }
    )
    # 查询DAG配置
    response = table.query(
        KeyConditionExpression=Key('dag_id').eq('mwaa_ddb_integration') & Key('config_key').eq('data_source')
    )
    print("DAG配置:", response['Items'][0]['config_value'])

with DAG('mwaa_ddb_hook_demo', start_date=datetime(2024,5,20), schedule_interval='@daily') as dag:
    config_task = PythonOperator(
        task_id='manage_dag_config',
        python_callable=use_ddb_hook_for_config
    )

注：MWAA默认已预装apache-airflow-providers-amazon包，无需额外安装。

4. 分场景替换S3策略 #

• 任务结果存储：替代S3的对象存储，用DynamoDB存储结构化结果，支持快速查询任务历史与状态，适合高频读写场景。
• DAG配置管理：将原存于S3的零散配置文件（JSON/YAML）迁移到DynamoDB，按DAG/任务维度存储，Airflow启动时直接读取，避免重复下载S3文件。
• 任务间数据传递：对于小体量结构化中间数据，用DynamoDB替代S3作为缓存层，减少IO延迟，提升任务串联效率。

5. 性能优化要点 #

• 合理设计表结构：根据查询模式设置复合主键（如task_id+execution_date）与二级索引，避免全表扫描。
• 开启自动扩缩容：为DynamoDB表开启读写吞吐量自动扩缩容，匹配MWAA任务并发量的动态变化，避免性能瓶颈。
• 混合存储方案：对于大体积非结构化数据，建议采用「DynamoDB存元数据+S3存文件」的模式，兼顾性能与存储成本。

内容的提问来源于stack exchange，提问作者Nyk Casseano