Cloud - Snowflake(2) 조작하기

프로젝트를 진행하면서 snowflake를 활용하여 데이터 웨어하우스 구축에 사용했던 것들을 모아서 정리하였습니다.

Colab을 활용하여 snowflake조작하기

: snowflake에 데이터를 colab으로 넣을 수 없을까 생각하다가 찾았던 라이브러리이다. 그러나 실사용에서는 snowflake의 웹을 통해서 넣는 것이 편해 잘 사용되지는 않았다.

• 설치

%pip install snowflake-connector-python
import snowflake.connector

 

•  계정 정보를 입력해서 연결 수행

ctx = snowflake.connector.connect(
    user='<user_name>',
    password='<password>',
    account='<account_name>'
    )
cs = ctx.cursor()
try:
    cs.execute("SELECT current_version()")
    one_row = cs.fetchone()
    print(one_row[0])
finally:
    cs.close()
ctx.close()

• SQL 입력

cs.cusor().execute(query)

 

추가 정보는 다음 블로그를 참고 : Snowflake Python에서 사용하는 법(Python Connector, Snowpark)

S3와 연결

: 연결과정의 경우 블로그의 글을 참고하며, 실제로 사용하였고, 동작한 코드위주로 정리글을 남긴다.

클라우드와 s3를 연결하기위해서는 3가지의 과정이 필요하다.

Integration → staging → COPY INTO

IAM Policy 구성

: S3폴더에 액세스 하려면 몇가지 권한이 필요하다.

• s3:GetBucketLocation
• s3:GetObject
• s3:GetObjectVersion
• s3:ListBucket
• s3:PutObject
• s3:DeleteObject

JSON코드로 정책을 생성하면 다음과 같다.

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
              "s3:PutObject",
              "s3:GetObject",
              "s3:GetObjectVersion",
              "s3:DeleteObject",
              "s3:DeleteObjectVersion"
            ],
            "Resource": "arn:aws:s3:::<bucket>/<prefix>/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:GetBucketLocation"
            ],
            "Resource": "arn:aws:s3:::<bucket>",
            "Condition": {
                "StringLike": {
                    "s3:prefix": [
                        "<prefix>/*"
                    ]
                }
            }
        }
    ]
}

에는 권한을 갖는 s3버킷을 입력하고, 에는 오브젝트의 경로를 입력한다.

Integration

: 클라우드의 키나 액세스 토큰같은 자격 증명을 전달하지 않고도 연결을 수행하게 해주는 개체이다. 이를 이용하면 자격 증명 정보를 코드에 넣지 않고도 클라우드와의 연결이 가능해진다.

-- integration
CREATE STORAGE INTEGRATION s3_connect
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = '<aws_role_arn>'
  STORAGE_ALLOWED_LOCATIONS = ('<s3_bucket_location>');

STAGING

: 파일을 snowflake로 로드하기 전에 임시로 저장하는 영역을 생성한다. 종류는 사용자, 테이블, 외부 stage가 있다.

-- staging
USE SCHEMA DATABASE_PROJECT.RAW_DATA;

CREATE STAGE my_s3_stage
  STORAGE_INTEGRATION = s3_connect
  URL = '<s3_bucket_location>';

COPY INTO

• load

-- 로드
COPY INTO SCHEMA.TABLE_NAME
FROM <파일 경로>
FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1); --첫 줄의 헤더를 스킵하는 포맷

• unload

-- CSV 형식으로 언로드
COPY INTO @my_ext_unload_stage/TABLE_NAME
FROM TABLE_NAME
FILE_FORMAT = (TYPE = 'CSV' FIELD_OPTIONALLY_ENCLOSED_BY = '"' FIELD_DELIMITER = ',' COMPRESSION = 'NONE');
-- SINGLE = TRUE; single로 하게되면 파일 하나로 언로드된다.

참고 및 출처) Amazon S3에 액세스하도록 Snowflake 설정하기