Glue에서 실패하는 job이 있다면 Lambda함수를 통해 Slack에 통지하기

대략적으로 순서를 나열해보자면 아래와 같다.

 

Glue job

AWS Glue란 대체 무엇인가

AWS Glue는 까다롭고 시간이 많이 소요되는 데이터 검색, 변환, 작업 일정 조정 등과 같은 작업은 간소화 및 자동화하는 종합 관리형 데이터 카탈로그 및 ETL(추출, 변환, 로드) 서비스라고 AWS에 소개되어있다.

 

ETL처리의 예를 들자면, Amazon Redshift에서 데이터를 unload(추출)해 Amazon S3에 보관하면서 csv에서 parquet으로 변환하고 Redshift로 load하는 등의 처리를 들 수 있다.

 

데이터 카탈로그에 대한 알기 쉬운 설명: Data Catalog 와 DCAT (Data Catalog Vocabulary) 개요

 

간단히 말하자면 데이터 카탈로그란 메타데이터(데이터에 대해 구조화 된 데이터)를 관리할 수 있게 하는 장치이다.

AWS Management Console을 사용하여 데이터 원본을 등록하면 AWS Glue가 데이터 원본을 크롤링한 후 JSON, CSV, Parquet 등 많이 사용되는 원본 형식 및 데이터 유형에 맞게 사전에 지정된 분류자를 사용하여 데이터 카탈로그를 구성한다.
AWS Glue는 Scala 또는 Python으로 ETL 처리 코드를 생성하여 원본으로부터 데이터를 추출하고 해당 데이터를 대상 스키마에 맞춰 변환하고, 이를 대상으로 로드한다.

Job과 Workflow

여기서 ETL 처리 스크립트를 Job으로 생성해 관련된 여러 Job을 이어 Workflow를 작성할 수 있다.
Amazon CloudWatch를 이용해 Workflow를 언제 어느 빈도로 실행시킬 것인지 정해주면 자동으로 정기적인 데이터처리가 가능해지는 것이다.

 

CloudWatch의 event

Amazon CloudWatch

먼저, Amazon CloudWatch란, AWS의 여러 서비스를 모니터링하는 서비스이다.
작업이 제대로 이루어지는지에 대한 log, error log가 여기에 기록된다.

 

위에서 설명한 AWS Glue에서 작성한 Job을 정기 실행했을 때 100% 성공만 한다고 할 수는 없다.
그 여부를 알 수 있는 것이 event이다.

Glue Job으로 발생한 event를 Lambda함수에게 전하기

Amazon CloudWatch에는 event rule이라는 것이 있다.
특정 event가 발생했을 때 트리거될 대상을 정해놓는 것이다.
이 단계에서 어떤 Lambda함수를 트리거할지 정할 수 있다.

콘솔에서 event rule 생성하기

규칙 생성 화면으로 가 아래와 같이 설정하면 모든 Glue Job event를 glueJobErrorNotification라는 Lambda함수에 전해줄 수 있게 된다.

cdk로 event rule 생성하기

Lambda함수를 CDK stack으로 생성하고 관리할 때, 그 stack 안에 아래와 같이 event rule을 생성하는 코드를 작성한다.

const glueJobErrorNotifyLambda = new lambda.Function(
        stack,
        'GlueJobErrorNotification',
        {
            code: lambda.Code.fromAsset(
                path.join(
                    __dirname,
                    'lambda/function/script/path'
                )
            ),
            functionName: 'GlueJobErrorNotification',
            handler: 'lambda_function.lambda_handler',
            timeout: cdk.Duration.seconds(300),
            runtime: lambda.Runtime.PYTHON_3_7,
            role: lambdaRoleArn,
            layers: [layers_for_import_in_the_lambda_function]
        }
    );

const glueJobErrorEvent = new event.Rule(stack, 'GlueJobErrorEvent', {
        targets: [new targets.LambdaFunction(glueJobErrorNotifyLambda)]
    });
    glueJobErrorEvent.addEventPattern({
        source: ['aws.glue']
    });

 

Lambda함수

먼저, AWS Lambda란 대체 무엇인가

AWS 서비스를 이용하면서 발생하는 event에 대해 특정 코드를 실행시키기 위한 서비스이다.
함수가 실행되었을 때만 요금이 발생한다.

 

일단 함수를 작성해 놓으면 event를 Amazon CloudWatch에 로그로써 기록하는 AWS 서비스를 통해 미리 작성해 놓은 함수를 실행시킬 수 있다.
즉, event로 Labmda함수가 트리거된다는 것이다.

event 외에도 Amazon Api Gateway를 통해 HTTP에 응답할 수도 있고 AWS SDK(python에서는 boto3라는 라이브러리)를 통해 API 호출로 트리거될 수도 있다.

 

예를 들어, AWS의 어떠한 데이터베이스에 새로운 데이터를 추가했다는 event가 Amazon CloudWatch에 기록되면, 새로 추가된 데이터를 처리하는 Labmda함수에 이를 전달해 자동으로 데이터를 처리하게 할 수 있다는 것이다.

event를 전해주는 것은 Amazon CloudWatch가 한다(물론 우리가 그렇게 할 수 있도록 지정해주긴 해야 한다. 자동으로 알아챌 수는 없지 않겠는가).

 

기본적인 Lambda함수 lambda_handler가 어떻게 생겼는지 보면 이해하기 좀 더 쉬울 것이다(event를 인수로 받고 있기 때문에).
참고로 context라는 것은 런타임 정보 즉 어떤 언어로 작성되었는지를 판단하기 위한 인수로, 실행시킬 때 따로 전해주는 것이 아니라 함수를 작성하는 단계에서 런타임을 고르기만 하면 알아서 인수를 받는다.

def lambda_handler(event, context):
    return

트리거 될 Lambda함수 작성하기

이번에 하고 싶은 것은 AWS Glue에서 실행시킨 Job의 결과를 담은 event를 전해받아 만약 Job이 실패했다면(상태가 FAILED, TIMEOUT, STOPPED라면) Labmda함수가 Slack에 이 상황을 통지하도록 하는 것이다.

 

Job의 실행 결과 정보를 담은 event는 아래와 같이 생겼다.
이것은 샘플 event인데 AWS CloudWatch에서 얻을 수 있다.

{
  "version": "0",
  "id": "sample_id",
  "detail-type": "Glue Job State Change",
  "source": "aws.glue",
  "account": "123456789012",
  "time": "sample_time",
  "region": "sample_region",
  "resources": [],
  "detail": {
    "jobName": "MyJob",
    "severity": "INFO",
    "state": "FAILED",
    "jobRunId": "sample_job_run_id",
    "message": "Job run failed"
  }
}

Job의 상태가 실패인지 성공인지를 알아야하기 때문에 event["detail"]["state"]를 읽어야한다.

def lambda_handler(event, context):
    if event["detail"]["state"] in ["FAILED", "STOPPED", "TIMEOUT"]:
        post_to_slack()
    ...

 

검증하기

검증 방법1: 샘플 event로 Lambda에서 test해보기

위에서 첨부한 샘플 event는 event rule 생성시에 아래에 만들어지므로 이것을 Lambda의 테스트 event로 사용하면 된다.
Lambda함수를 만들어 저장을 누르고 그 옆에 test 버튼이 있을텐데 그 버튼 왼쪽에 event를 지정할 수 있는 란이 있다.

Configure test event를 눌러 새로운 event를 아래와 같이 생성한다.

test버튼을 누르고 Slack에 제대로 메시지가 가는지 확인한다.

검증 방법2: 실제로 job을 만들어 일부러 실패시키기

실제로 ETL처리를 하는 것이 아닌 무조건 실패하는 python 스크립트를 만들어 Job으로 만든다.

true_bool = False
assert true_bool == True 

위에서 설명한 것 처럼 Amazon CloudWatch에서 event rule을 생성한 뒤 job을 실행시키고 Slack에 제대로 메시지가 가는지 확인한다.