使用Airflow CreateEMRJobFlowOperator创建带自动终止策略的EMR集群报错

问题详情 #

尝试通过Airflow DAG创建带有自动终止空闲超时设置的EMR集群时，系统返回参数验证错误，提示AutoTerminationPolicy是未知参数。

报错信息 #

raise ParamValidationError(report=report.generate_report())
botocore.exceptions.ParamValidationError: Parameter validation failed:
Unknown parameter in input: "AutoTerminationPolicy", must be one of: Name, LogUri, LogEncryptionKmsKeyId, AdditionalInfo, AmiVersion, ReleaseLabel, Instances, Steps, BootstrapActions, SupportedProducts, NewSupportedProducts, Applications, Configurations, VisibleToAllUsers, JobFlowRole, ServiceRole, Tags, SecurityConfiguration, AutoScalingRole, ScaleDownBehavior, CustomAmiId, EbsRootVolumeSize, RepoUpgradeOnBoot, KerberosAttributes, StepConcurrencyLevel, ManagedScalingPolicy, PlacementGroupConfigs

原配置代码片段 #

JOB_FLOW_OVERRIDES = {
    'Name': 'X',
    'ReleaseLabel': "{{ReleaseLabel}}",
    "Applications": [{"Name": "Hadoop"}, {"Name": "Spark"}, {"Name": "Hive"}],
    "AutoTerminationPolicy": { 
      "IdleTimeout": 60 * 10
      },
    'LogUri': LogUri,
    'Instances': {
        'Ec2SubnetId': "{{Ec2SubnetId}}",
        'EmrManagedMasterSecurityGroup': "{{EmrManagedMasterSecurityGroup}}",
        'ServiceAccessSecurityGroup': "{{ServiceAccessSecurityGroup}}",
        'EmrManagedSlaveSecurityGroup': "{{EmrManagedSlaveSecurityGroup}}",
        'InstanceGroups': [
        # 后续配置省略
    }
}

解决办法 #

• AutoTerminationPolicy不是顶层参数，必须嵌套在Instances字段内
• 修改配置，将AutoTerminationPolicy移动到Instances字典中

修改后的代码示例 #

JOB_FLOW_OVERRIDES = {
    'Name': 'X',
    'ReleaseLabel': "{{ReleaseLabel}}",
    "Applications": [{"Name": "Hadoop"}, {"Name": "Spark"}, {"Name": "Hive"}],
    'LogUri': LogUri,
    'Instances': {
        'Ec2SubnetId': "{{Ec2SubnetId}}",
        'EmrManagedMasterSecurityGroup': "{{EmrManagedMasterSecurityGroup}}",
        'ServiceAccessSecurityGroup': "{{ServiceAccessSecurityGroup}}",
        'EmrManagedSlaveSecurityGroup': "{{EmrManagedSlaveSecurityGroup}}",
        'AutoTerminationPolicy': { 
            "IdleTimeout": 60 * 10
        },
        'InstanceGroups': [
        # 后续配置省略
    }
}

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