Lambda 도입 개요

데일리로 새로운 bs4를 통해 데이터를 스크래핑하여 parquet 형태로 AWS S3에 적재하는 프로그램을 생성했다. 기존에는 모든 프로그램을 AWS의 EC2로 Crontab을 설정하여 스케쥴링(Scheduling)했었는데, 이번 경우 아래와 같은 이유로 AWS Lambda 함수를 이용하기로 했다.

비용: AWS Lambda는 실행 시간에 따라 과금된다. EC2 인스턴스를 사용하면 인스턴스를 실행하는 데 필요한 시간 동안 비용이 발생한다. 따라서 15분 이내 돌아가는 프로그램을 실행하는 경우(Lambda는 실행 시간이 15분을 초과 할 수 없다), Lambda 함수를 사용하면 EC2 인스턴스를 사용하는 것보다 훨씬 저렴하다.
관리: Lambda 함수를 사용하면 EC2 인스턴스를 사용하는 경우보다 서버 관리에 필요한 노력과 시간을 절약할 수 있다. Lambda 함수는 AWS에서 관리하기 때문에 배포, 확장, 모니터링 및 유지 관리를 AWS가 담당한다.
확장성: Lambda 함수를 사용하면 트래픽이 증가할 때 자동으로 확장된다. EC2 인스턴스를 사용하는 경우, 인스턴스를 추가하고 구성해야한다.
빠른 시작: Lambda 함수를 사용하면 코드를 쉽게 작성하고 테스트할 수 있다. 또한 AWS 콘솔에서 함수를 만들고 구성할 수 있으므로 EC2 인스턴스를 프로비저닝하고 구성하는 것보다 더 빠르게 시작할 수 있다.

이 포스팅은 람다 함수를 도입하며 발생했던 문제 위주로만 작성했다.

데이터 수집 예제 코드

필자는 여러 프로그램에 공유하는 함수가 포함된 파이썬 프로그램인 모듈(module)이 있었기 때문에 이 상황을 가정하여 절차를 설명할 것이다. 아래는 차례대로 로컬에서 디버깅을 마친 모듈과 수집을 진행하는 프로그램의 예제 코드이다.

module.py

import boto3
import requests
from bs4 import BeautifulSoup
import pandas as pd

s3 = boto3.client("s3")

def crawl_data(url):
    res = requests.get(url)
    soup = BeautifulSoup(res.content, "html.parser")
    ... # 크롤링 코드
    df = pd.DataFrame(...) # 크롤링 결과를 데이터프레임으로 변환
    return df

def save_to_s3(df, bucket, key):
    csv_buffer = df.to_csv(index=False).encode()
    s3.put_object(Body=csv_buffer, Bucket=bucket, Key=key)

crawler.py

def main():
    url = "..." # 크롤링할 URL
    df = crawl_data(url)
    bucket = "..." # S3 버킷 이름
    key = "..." # S3 객체 키
    save_to_s3(df, bucket, key)

if __name__ == '__main__':
    main()

Lambda 함수 생성

AWS Lambda 함수를 생성한다. 비교적 간단하다.

AWS Management Console에 로그인한다.
Lambda 서비스로 이동한다.
“함수 생성(Create Function)” 버튼을 클릭한다.
“새로 작성(Author from scratch)”을 선택한다.
“함수 이름(Function name)”을 입력한다.
- “myTestFunction”
“런타임(Runtime)”에서 본인과 같은 버전의 “Python 3.n”을 선택한다.
“함수 생성(Create function)” 버튼을 클릭한다.

`lambda_function.py`에 코드 업로드

`lambda_handler` 추가

AWS Lambda 함수를 만들 때, 해당 함수는 항상 “lambda_handler”라는 이름의 진입점(entry point)을 가져야 한다. 이 진입점은 Lambda가 함수를 실행할 때 호출되며, 함수가 수행해야 할 코드를 정의한다.

lambda_handler 함수는 특정 형식으로 작성되어야 하며, 일반적으로 다음과 같은 형식을 가지고 있다.

def lambda_handler(event, context):
    # 함수가 수행해야 할 코드 작성

따라서 crawler.py에 아래와 같은 코드를 추가하여 작성하면 된다.

def main():
    url = "..." # 크롤링할 URL
    df = crawl_data(url)
    bucket = "..." # S3 버킷 이름
    key = "..." # S3 객체 키
    save_to_s3(df, bucket, key)

def lambda_handler(event, context): # 람다 실행 시 동작
    main()
    return {
        'statusCode': 200,
        'body': json.dumps('Successfully executed main function!')
    }

if __name__ == '__main__': # 로컬 실행 시 동작
    main()

여기서 “event”는 Lambda 함수에 전달되는 이벤트 데이터이며, “context”는 실행 컨텍스트를 나타내는 객체이다. 이들 매개변수는 Lambda 함수 실행 시 자동으로 전달되므로, 함수 코드에서 직접 이들 매개변수를 지정할 필요는 없다.

위와 같이 추가한다면 진입점으로 작동하게 되며, 정상적으로 lambda 함수가 실행될 것이다. if __name__ == '__main__':과 lambda_handler를 동시에 두어도, 각각의 환경마다 무시할 것이기에 굳이 수정할 필요없이 필자는 로컬의 코드와 람다 함수의 코드를 통일시켰다.

위와 같은 코드를 lambda_function.py에 업로드하면 되는데 여러가지 방법이 있다.

작성한 코드를 패키징하여(.zip) 업로드
Amazon S3에 저장 후 업로드
직접 copy&paste

모두 비교적 간단하며 이 포스팅에서는 넘어가겠다.

계층(Layer) 추가

레이어(Layer)는 Lambda 함수에서 공유할 수 있는 코드, 라이브러리 및 기타 리소스의 패키지이다. 이를 통해 여러 Lambda 함수에서 동일한 코드 및 라이브러리를 재사용할 수 있다. 레이어는 별도의 파일 시스템으로 분리되어 관리되며, Lambda 함수에서 필요한 경우 레이어를 호출하여 사용할 수 있다.

앞서 설명했 듯 필자도 여러가지 함수에서 사용할 모듈이 있다고 했고, 파이썬 프로그램에서 사용한 각종 라이브러리(boto3, bs4)를 패키징하여 등록하여야 정상적으로 실행시킬 수 있다.

사용할 모듈 준비

일단 이 쯤에서 가장 중요한 정보를 짚고 넘어간다. 람다에서는 우리가 로컬에서 사용한 pandas를 그대로 사용할 수 없다. 다행히 람다에서 pandas 레이어를 기본적으로 제공하니, 우리가 필요한 모듈을 모두 준비한 뒤 다시 알아보자.

`python` 폴더 생성

먼저 python이라는 이름의 빈 폴더를 하나 만들어 준다.Python Runtime에서 폴더명은 꼭 python이어야 한다.

$ mkdir python

폴더 내 라이브러리 생성

앞에서 말했듯, 판다스는 제외하고 설치한다.

$ pip install -t python boto3 requests bs4 # 코드에서 사용된 라이브러리

폴더 안의 *.dist-info와 __pycache__를 제거한다.

$ cd python
python$ rm -r *.dist-info __pycache__
python$ cd ..
$

커스텀 모듈 복사

필자는 필자가 만든 공유할 커스텀 모듈이 있는 상태이다. 여러 함수에 공유하기 위해 이 또한 모듈 폴더에 포함시켜준다. 커스텀 모듈이 없을 경우 넘어가도 좋다.

$ mv module.py python

모듈 폴더 압축

$ zip -r python.zip python

이제 디렉토리에 python.zip가 생성되었다.

계층 생성

Lambda > 계층에서 “계층 생성”을 한다.
이름을 작성한다.
- myTestModule
앞서 압축한 python.zip파일을 업로드한다.
1. .zip파일 직접 업로드
2. python.zip S3에 저장 후 링크 URL을 통한 업로드
“생성”을 눌러 생성한다.

함수 내 계층 추가

함수 > 계층에서 “[Add a layer]”를 누른다.
“ARN 지정”을 누른 뒤, 아까 전에 추가한 계층 ARN을 버전에 맞게 추가하여 “확인”을 누른다.

판다스 사용하는 경우

만약 함수에서 판다스를 이용하는 경우, AWS에서 기본적으로 제공하는 계층을 추가하면 된다.

함수 > 계층에서 “[Add a layer]”를 누른다.
“AWS 계층”을 누른 뒤, “AWSSDKPandas-Python”을 선택하여 추가한다.

테스트 및 실행

자 이제 모든 준비는 끝났다. 테스트 및 실행을 해보자. 그 전에 기본 실행 시간이나 메모리가 굉장히 작기 때문에 구성을 잠깐 손봐야한다.

구성 편집

해당 함수 > 구성 > 일반 구성에서 “편집”을 누른다.
해당 함수의 코드가 충분히 실행될 만큼의 메모리와 실행 제한 시간을 알맞게 변경한다.
- 사용 메모리와 시간에 따라서 부과되는 요금이 다르기 때문에, 이 부분을 최적화하기 전략도 존재하니 따로 알아보면 좋다.
“저장”을 누른다.

테스트

코드를 최종적으로 저장한 뒤 “Deploy” 버튼을 눌러 배포 후 “Test”를 진행한다.
“Event name”을 입력하고 “Create” 버튼을 클릭한다. (이벤트는 빈 JSON 객체 {}로 생성 가능하고, 굳이 아무것도 변경하지 않아도 된다.)
정상 동작을 확인한다.

스케쥴링

동작이 정상적인 람다 함수를 스케쥴링할 방법은 많다. 필자는 에어플로우(Airflow)를 활용하여 스케쥴링 했지만, 이 포스팅에서 비교적 간단한 이벤트(EventBridge)만을 간략히 설명할 것이다.

해당 함수에서 “트리거 추가(Add trigger)” 버튼을 클릭한다.
“EventBridge(CloudWatch Events)”를 선택한다.
“새 규칙 생성(Create a new rule)”을 선택한다.
“규칙 이름(Rule name)”을 입력한다.
“예약 표현식(Schedule expression)”에 스케줄 표현식을 입력한다.
- (예: “cron(0 12 * * ? *)”은 매일 정오에 실행)
“추가(Create)” 버튼을 클릭한다.

이제 프로그램이 람다에서 자동으로 실행될 것이다.

번외

필자는 이와 같은 스크래핑 프로그램이 3개가 존재했고, 2개는 람다 함수화 시켜 airflow에 성공적으로 dag로 스케쥴링 했지만 하나의 프로그램은 실행 시간이 15분이 넘어가서 람다 함수로 실행할 수 없는 일이 발생했다. 결국 그 프로그램을 위해서 EC2를 연결하여 SSHOperators를 통한 스케쥴링을 진행했는데, 그 과정은 다음 포스팅에서 다룰 것이다.