Snowflake Connector(Python)용 AWS Lambda Custom Layer

Published in

Cloud Villains

6 min readJul 8, 2024

Snowflake의 Task를 AWS Lambda를 이용하면 필요한 조건에 따라서 실행할 수 있습니다. 그렇지만 AWS Lambda에서는 Snowflake Connector를 기본적으로 지원하지 않기 때문에 Custom Layer를 생성해서 함수가 잘 작동할 수 있도록 환경을 구성해야 합니다. 따라서 오늘은 Snowflake Connector(Python)용 AWS Lambda Custom Layer를 아래의 순서로 생성해보겠습니다.

미리 알아두어야 할 사항
리눅스 환경에서 파일을 생성하기 위해서 Ubuntu EC2를 생성
Dependencies 설치 및 압축 파일 만들기
AWS Lambda Custom Layer 생성
Sample 코드 실행으로 정상적인 동작 확인하기

미리 알아두어야 할 사항

Python용 AWS Lambda Custom Layer는 Amazon Linux에서 동작합니다. 그래서 리눅스 환경에서 Zip File을 생성해야 합니다. 특히 개인 랩탑(Mac, Windows)에서 Zip File 생성 시 잘 동작하지 않습니다. 현시점(2024–07–08) 기준으로 Snowflake는 Python 버전 3.11을 추천합니다.

리눅스 환경에서 파일을 생성하기 위해서 Ubuntu EC2를 생성

인스턴스에서 Launch Instances를 클릭합니다

이름을 넣고 우분투를 선택하고 key pair를 생성합니다.

Connect to your instance를 클릭합니다.

아래의 방법을 참고하여 생성된 EC2 인스턴스에 SSH를 통해서 접속합니다.

Dependencies 설치 및 압축 파일 만들기

접속한 SSH Client(Terminal)에서 아래의 코드를 실행하여 Dependency를 설치합니다.

# sudo apt를 업데이트합니다
sudo apt update

# python dependency를 설치하기 위해 pip를 설치합니다
sudo apt install pip

# 압축파일을 만들기 위해 zip을 설치합니다.
sudo apt install zip

# 디렉토리를 만들고 snowflake connector를 설치합니다.
mkdir -p temp/python
cd temp/python
pip3 install snowflake-connector-python -t .

압축 파일 만들기

# zip 파일을 생성합니다
cd ..
zip -r9 ../snowflake-connector-python.zip .

로컬 폴더에 압축 파일 다운로드하기

# 새로운 터미널에서 아래의 명령어를 실행합니다
# pem키, EC2 디렉토리, 로컬 PC 디렉토리는 사용자의 상황에 맞게 변경합니다
scp -i "snowflake_connector_ec2_key.pem" ubuntu@ec2-3-36-67-138.ap-northeast-2.compute.amazonaws.com:/home/ubuntu/snowflake-connector-python.zip /Users/mzc01-ssm/Downloads snowflake-connector-python.zip

AWS Lambda Custom Layer 생성

Layers를 선택하고 Create layer를 클릭합니다.

이름을 기입하고 zip file을 업로드하고 생성합니다. 생성 후 ARN을 복사합니다.

Sample 코드 실행으로 정상적인 동작 확인하기

함수를 생성합니다.

함수 이름, 런타임을 설정하고 함수를 생성 완료합니다.

Add a layer를 클릭하고 복사 된 ARN으로 Layer를 더합니다.

아래의 함수를 넣고 Deploy를 누릅니다.

import json
import snowflake.connector

conn = snowflake.connector.connect(
user='xxxxxxxxxxxxx',
password='xxxxxxxxxxxxx',
account='xxxxxxxxxxxxx',
warehouse='xxxxxxxxxxxxx',
database='xxxxxxxxxxxxx',
schema='xxxxxxxxxxxxx'
)
cs = conn.cursor()
cs.execute('execute task mzc_test_task1')
one_row = cs.fetchone()

cs.close()
conn.close()
    
def lambda_handler(event, context):

    # TODO implement
    return {
        'statusCode': 200,
        'SnowflakeTaskStatus': one_row[0]
    }