Amazon MWAA에서 Amazon EMR 단계를 트리거할 때 발생하는 오류를 해결하려면 어떻게 해야 합니까?

간략한 설명

Amazon MWAA에서 Amazon EMR 단계를 트리거할 때 다음과 같은 오류가 발생할 수 있습니다.

"An error occurred (AccessDeniedException) when calling the DescribeCluster operation"

"An error occurred (AccessDeniedException) when calling the AddJobFlowSteps operation"

"EMR endpoint is not reachable: botocore.exceptions.ConnectTimeoutError: Connect timeout on endpoint URL"

"An error occurred (InvalidRequestException) when calling the DescribeCluster operation: Cluster id is not valid."

"An error occurred (ValidationException) when calling the AddJobFlowSteps operation: Maximum number of active steps(State = 'Running', 'Pending' or 'Cancel_Pending') for cluster exceeded."

"An error occurred (ValidationException) when calling the AddJobFlowSteps operation: jobFlowId is not valid."

"An error occurred (ValidationException) when calling the AddJobFlowSteps operation: A job flow that is shutting down, terminated, or finished may not be modified."

"No module named 'airflow.providers.amazon.aws.operators.emr_add_steps'"

해결 방법

참고: AWS Command Line Interface(AWS CLI) 명령 실행 시 오류가 발생하는 경우, AWS CLI 오류 문제 해결을 참조하십시오. 또한 최신 AWS CLI 버전을 사용하고 있는지 확인하십시오.

Amazon EMR API 호출 중에 AccessDeniedException 오류 발생

이러한 오류는 Amazon MWAA 런타임 역할에 Amazon EMR 작업과 관련된 권한 문제가 있을 때 발생합니다. 이러한 오류를 해결하려면 Amazon MWAA 런타임 역할에 DescribeCluster 및 AddJobFlowSteps 권한이 있어야 합니다.

예:

{
"Version": "2012-10-17",
"Statement": [
    {
        "Sid": "EMRStepPermissions",
        "Effect": "Allow",
        "Action": [
                "elasticmapreduce:DescribeCluster",
                "elasticmapreduce:AddJobFlowSteps"
                ],
        "Resource": "arn:aws:elasticmapreduce:example-region:example-account-id:cluster/example-cluster-id"
    }
    ]
}

참고: example-region을 AWS 리전으로, example-account-id를 계정 ID로, example-cluster-id를 클러스터 ID로 바꾸십시오.

EMR endpoint is not reachable: botocore.exceptions.ConnectTimeoutError: Connect timeout on endpoint URL

이 오류는 잘못된 네트워킹 구성으로 인해 Amazon MWAA 환경이 Amazon EMR 클러스터에 도달할 수 없을 때 발생합니다. 이 오류를 해결하려면 Amazon MWAA Amazon Virtual Private Cloud(Amazon VPC) 구성이 클러스터로의 아웃바운드 트래픽을 허용하는지 확인하십시오.

퍼블릭 라우팅의 경우 보안 그룹과 서브넷에 클러스터로의 아웃바운드 트래픽을 허용하는 규칙과 경로가 있는지 확인합니다. 프라이빗 라우팅의 경우 Amazon EMR 엔드포인트가 Amazon MWAA 환경 서브넷 및 보안 그룹과 연결되어 있는지 확인합니다.

Amazon EMR이 다른 Amazon VPC에 있는 경우 Amazon VPC 피어링을 사용하여 적절한 네트워크 연결을 설정합니다.

An error occurred (InvalidRequestException) when calling the DescribeCluster operation: Cluster id is not valid

이 오류를 해결하려면 Amazon MWAA 환경이 구성된 것과 동일한 리전 및 계정에 클러스터 ID가 있는지 확인하십시오. 또한 AWS Identity and Access Management(IAM) 사용자가 클러스터 세부 정보를 확인할 수 있도록 하려면 JOB_FLOW_OVERRIDES 파라미터에서 VisibleToAllUsers를 True로 설정합니다. 자세한 내용은 요청 파라미터를 참조하십시오.

Amazon MWAA 런타임 역할이 클러스터를 나열하는 데 필요한 권한을 확인하려면 describe-cluster 명령을 실행합니다.

Amazon EMR 클러스터에 대한 추가 세부 정보를 보려면 Amazon EMR 클러스터 상태 및 세부 정보 보기를 참조하십시오.

An error occurred (ValidationException) when calling the AddJobFlowSteps operation: jobFlowId <'some-string'> is not valid

이 오류는 EmrAddStepsOperator에서 잘못된 job_flow_id가 전달될 때 발생합니다. 이 문제를 해결하려면 Amazon EMR 클러스터 ID를 job_flow_id로 사용해야 합니다. 자세한 내용은 Apache Airflow 문서의 파라미터를 참조하십시오.

An error occurred (ValidationException) when calling the AddJobFlowSteps operation: A job flow that is shutting down, terminated, or finished may not be modified

이 오류는 Amazon EMR 작업 흐름이 중지, 종료 또는 완료되어 AddJobFlowSteps 작업을 수행할 수 없을 때 발생합니다. 이 오류를 해결하려면 AWS CloudTrail을 사용하여 작업 로그에서 클러스터가 종료된 시간과 EmrAddStepsOperator 작업이 시작된 시간을 확인하십시오. 클러스터를 시작하는 작업을 시작하기 전에 클러스터 종료 명령을 트리거한 경우 오류가 발생합니다.

이 오류를 해결하려면 DAG 코드의 JOB_FLOW_OVERRIDES 파라미터에 KeepJobFlowAliveWhenNoSteps를 True로 추가하십시오. 이 설정은 단계가 완료된 후 클러스터를 대기 중 상태로 전환합니다.

예:

JOB_FLOW_OVERRIDES = {
    "Name": "Data-Pipeline-" + execution_date,
    "ReleaseLabel": "emr-5.29.0",
    "LogUri": "s3://{}/logs/emr/".format(S3_BUCKET_NAME),
    "Instances": {
        "InstanceGroups": [
            {
                "Name": "Master nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "MASTER",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 1
            },
            {
                "Name": "Slave nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "CORE",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 2
            }
        ],
        "TerminationProtected": False,
        "KeepJobFlowAliveWhenNoSteps": True
    }
}

단계가 완료된 후 클러스터를 종료하려면 다음 방법 중 하나를 사용합니다.

EmrTerminateJobFlowOperator를 사용하여 단계 완료 시 클러스터를 종료하는 DAG의 마지막 작업을 생성합니다.

cluster_remover = EmrTerminateJobFlowOperator(
        task_id='remove_cluster',
        job_flow_id=cluster_creator.output,
        aws_conn_id='aws_default',
    )

또는 지정된 유휴 시간이 지나면 클러스터를 종료하는 자동 종료 정책을 JOB_FLOW_OVERRIDES에 설정합니다. 다음 예시 구성은 60분의 유휴 시간이 지나면 클러스터를 종료합니다.

JOB_FLOW_OVERRIDES = {
    "Name": "Data-Pipeline-" + execution_date,
    "ReleaseLabel": "emr-<version>",
    "LogUri": "s3://{}/logs/emr/".format(example-s3-bucket-name),
    "Instances": {
        "InstanceGroups": [
            {
                "Name": "Master nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "MASTER",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 1
            },
            {
                "Name": "Slave nodes",
                "Market": "ON_DEMAND",
                "InstanceRole": "CORE",
                "InstanceType": "m5.xlarge",
                "InstanceCount": 2
            }
        ],
        "TerminationProtected": False,
        "KeepJobFlowAliveWhenNoSteps": True
    },
    "AutoTerminationPolicy": {"IdleTimeout": 3600}
}

자세한 내용은 Apache Airflow 웹 사이트의 작업 정의를 참조하십시오.

다른 JOB_FLOW_OVERRIDES를 보려면 run_job_flow를 참조하십시오.

Amazon EMR 자동 종료에 대한 자세한 내용은 Amazon EMR 클러스터 정리를 위한 자동 종료 정책 사용을 참조하십시오.

"No module named 'airflow.providers.amazon.aws.operators.emr_add_steps'"

Amazon MWAA 환경과 Amazon EMR 클러스터를 서로 다른 리전에 두어야 하는 경우 연결 ID를 하나 더 생성하십시오. 클러스터를 시작하려는 AWS 리전을 사용해야 합니다. 기본 리전을 변경하려면 다음 단계를 완료하십시오.

1. Apache Airflow UI를 엽니다.
2. 위쪽 탐색 창에서 관리자를 선택하고 연결을 선택한 다음 + 버튼을 선택합니다.
3. DAG에서 사용할 연결 ID의 이름을 입력합니다.
4. 연결 유형으로 Amazon Web Service를 선택합니다.
5. 텍스트 필드에 다음 JSON을 추가합니다.

   {"region_name": "example-region"}

   참고: example-region을 기본 리전으로 사용하려는 리전으로 바꾸십시오.
6. Amazon MWA에서 사용되는 각 Amazon EMR 운영자에 대해 새 연결 ID를 사용해야 합니다.

DAG 가져오기 오류

DAG 가져오기 오류가 발생하면 필요한 패키지가 설치되지 않았거나, 가져오기 경로가 다르거나, 운영자 이름이 잘못된 것입니다. 이러한 오류를 해결하려면 Amazon MWAA 환경에 어떤 apache-airflow-providers-amazon 패키지 버전이 설치되어 있는지 확인하십시오. 이를 위해서는 Airflow UI 공급자 페이지를 확인하십시오. 또는 Amazon MWAA 환경에 설치된 Apache Airflow 공급자 패키지를 참조하십시오. 그런 다음 공급자 패키지 버전에 따라 Airflow 공급자 문서에서 가져오기 경로와 운영자 이름을 확인합니다. 예를 들어, 아마존 MWAA 2.2.2에는 apache-airflow-providers-amazon 2.4.0이 있습니다. 버전 2.4.0의 가져오기 경로와 운영자 이름을 확인하려면 Apache Airflow 웹 사이트의 airflow.providers.amazon.aws를 참조하십시오.

각 Amazon EMR 운영자의 가져오기 경로는 다음과 같습니다.

• EmrAddStepsOperator: airflow.providers.amazon.aws.operators.emr_add_steps
• EmrCreateJobFlowOperator: airflow.providers.amazon.aws.operators.emr_create_job_flow
• EmrTerminateJobFlowOperator: airflow.providers.amazon.aws.operators.emr_terminate_job_flow

참고: apache-airflow-providers-amazon 8.16.0이 있는 Amazon MWAA 2.8.1의 경우, 모든 아마존 EMR 운영자가 공통 아마존 EMR 클래스로 이전되었습니다. 자세한 내용은 Apache Airflow 웹 사이트의 airflow.providers.amazon.aws.operators.emr을 참조하십시오.

아마존 EMR 운영자의 경로를 포함하려면 airflow.providers.amazon.aws.operators.emr에서 경로를 가져옵니다.

from airflow.providers.amazon.aws.operators.emr import EmrAddStepsOperator,
EmrCreateJobFlowOperator, EmrTerminateJobFlowOperator

참고: 사용 사례에 따라 DAG 가져오기 문을 수정해야 합니다.