如何通过MWAA Airflow DAG克隆EMR集群

完全可以，不需要更换Operator，也不需要申请新建集群的IAM权限，只要正确配置JOB_FLOW_OVERRIDES参数即可匹配你现有的克隆集群权限，完成运行中/已终止EMR集群的克隆操作。

核心原理 #

• EMR的克隆集群操作底层对应的就是RunJobFlow接口，当传入的配置和源集群配置完全一致、且调用方持有对应源集群的elasticmapreduce:CloneCluster权限时，AWS不会将请求判定为普通新建集群请求，会直接走克隆流程，不需要你持有全量的新建集群权限。
• 你之前找到的新建集群示例，本质是自定义JOB_FLOW_OVERRIDES的全量配置，你只要把这个参数的值替换为从源集群导出、并剔除无效字段后的配置，就能实现克隆效果。

具体实现步骤 #

• 第一步：拉取源集群配置。在DAG中通过EmrHook调用boto3的describe_cluster接口，传入待克隆的源集群ID（不管集群是运行中还是已终止状态都支持），拿到源集群的完整配置。
• 第二步：清洗配置字段。剔除所有EMR自动生成、不允许传入RunJobFlow接口的运行时字段，包括集群ID、ARN、状态信息、DNS地址、运行时长统计、自动生成的实例属性等，避免触发参数校验错误。
• 第三步：传入Operator执行。将清洗完成的配置作为JOB_FLOW_OVERRIDES参数传入EmrCreateJobFlowOperator即可，不需要额外配置其他参数。

参考实现代码：

from airflow.providers.amazon.aws.operators.emr import EmrCreateJobFlowOperator
from airflow.providers.amazon.aws.hooks.emr import EmrHook
from airflow.operators.python import PythonOperator
from airflow import DAG
from datetime import datetime

def get_cloned_cluster_config(**context):
    emr_hook = EmrHook(aws_conn_id="your_aws_connection_id")
    client = emr_hook.get_conn()
    # 替换为实际要克隆的源集群ID
    source_cluster_id = "j-XXXXXXXXXXXX"
    resp = client.describe_cluster(ClusterId=source_cluster_id)
    cluster_config = resp["Cluster"]
    
    # 剔除RunJobFlow接口不接受的运行时自动生成字段
    invalid_runtime_keys = [
        "Id", "ClusterArn", "Status", "StatusTimeline",
        "MasterPublicDnsName", "NormalizedInstanceHours",
        "ClusterType", "RequestedAmiVersion", "RunningAmiVersion"
    ]
    for key in invalid_runtime_keys:
        cluster_config.pop(key, None)
    
    # 可按需修改少量非核心配置，比如克隆后的集群名称
    cluster_config["Name"] = "cloned-emr-cluster-from-mwaa"
    return cluster_config

with DAG(
    dag_id="emr_clone_workflow",
    start_date=datetime(2024, 1, 1),
    schedule_interval=None,
    catchup=False
) as dag:
    fetch_source_config = PythonOperator(
        task_id="fetch_source_cluster_config",
        python_callable=get_cloned_cluster_config
    )

    create_cloned_cluster = EmrCreateJobFlowOperator(
        task_id="create_cloned_emr_cluster",
        aws_conn_id="your_aws_connection_id",
        job_flow_overrides="{{ ti.xcom_pull(task_ids='fetch_source_cluster_config') }}",
        region_name="cn-north-1" # 替换为实际使用的AWS区域
    )

    # 后续可直接串联你已经在使用的EmrAddStepsOperator提交Spark任务
    fetch_source_config >> create_cloned_cluster

注意事项 #

• 如果克隆时提示权限不足，先检查你是不是修改了源集群的核心配置（比如实例规格、VPC配置、安全组、EMR版本、预装应用集合、引导操作脚本等），一旦核心配置和源集群不一致，请求就会被判定为新建集群，触发权限拦截。
• 克隆已终止集群时不需要额外调整参数，只要拉取到的源集群配置完整，和克隆运行中集群的逻辑完全一致。

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