AWS Wrangler를 사용해서 외부데이터를 Redshift에 적재

AWS Wrangler?

AWS Wrangler는 AWS 서비스와 통합하여 데이터를 처리하고 관리하는 데 유용한 Python 라이브러리입니다. 특히, AWS 데이터 서비스(예: Amazon S3, Amazon Athena, Amazon Redshift, Amazon RDS 등)와 함께 사용할 때 매우 유용합니다. AWS Wrangler는 Pandas와 같은 데이터 처리 라이브러리와의 통합을 제공하여, 대규모 데이터 세트에 대한 작업을 간편하게 수행할 수 있습니다.

Connection 생성

Wrangler를 사용해서 레드시프트에 연결을 생성하는 방법으로는 glue와 secret manager 중 한가지를 선택해서 연결해야 한다.

Glue connection
AWS Glue Console → Connections → Create connection → Redshift 선택
이후 username, db name과 같은 필요 값을 기입하고 생성하면 된다.
Glue Connection은 Database 별로 생성해야 하므로 하나의 클러스터에서 여러개의 DB를 쓰는경우에는 여러 connection 생성이 필요 (wrangler 에서 connection 메소드 사용시에 db를 connection에 등록된 것과 다른 것으로 설정 할 수도 있긴 하다.)
Secret Manager
Secret Manager에서 암호를 생성하되 아래와 같은 구조의 secret을 가지면 된다.

  {
	  "host":"my-host.us-east-1.redshift.amazonaws.com",
	  "username":"test",
	  "password":"test",
	  "engine":"redshift",
	  "port":"5439",
	  "dbname": "mydb"
  }

dbname은 필수값은 아니며, connect 메소드 사용시에 지정해도 된다.
직접 위와같은 구조로 시크릿을 생성해도 되고, 콘솔에서 시크릿 생성시에 “Credentials for Amazon Redshift data warehouse” 시크릿 타입을 선택하면 필요한 값을 지정해주기 때문에 좀 더 간편하게 설정이 가능하다. (결과적으로는 아래와 같은 시크릿이 생성된다.)

{
	"username":"test",
	"password":"test",
	"engine":"redshift",
	"host":"test.host.ap-northeast-2.redshift.amazonaws.com",
	"port":5439,
	"dbClusterIdentifier":"test-cluster"
}

데이터 Redshift에 Load

wrangler를 사용해서 데이터를 Redshift에 적재하는 방법으로는 parquet 파일을 읽어와서 적재하거나, pandas 데이터프레임을 가지고 레드시프트에 적재하도록 할 수 있다.

Redshift connection 생성

import awswrangler as wr

wr_con = wr.redshift.connect(
	# connection이나 secret_id 두개 중 한개는 필수로 입력
	connection="glue_catalog_connection_name"
	secret_id="secret_manager_id",
	dbname=target_db_name
)

판다스 DF를 Load - 기본

wr.redshift.to_sql(
    df=df,  #적재하고자 하는 판다스 데이터프레임
    con=wr_con,  # 앞에서 생성한 커넥션
    schema=redshift_schema,  #레드시프트 테이블 스키마
    table=table_name,  # 레드시프트 테이블 이름
    mode=copy_mode,  # ["append", "overwrite", "upsert"] default: "append"
    overwrite_method=overwrite_method,  # ["drop", "cascade", "truncate", "delete"] default: "drop"
)

판다스 DF를 Load - 대용량

copy를 사용하게 되면, 라이브러리 내부적으로 S3에 stage Parquet 파일을 생성해서 이를 가져다가 레드시프트 테이블에 copy command를 사용하게 되어 있다.

공식문서에는 아래와 같은 안내사항이 기재되어 있는데, 높은 처리량과 높은 latency를 갖는 대용량의 데이터 프레임에 insert의 대안으로 copy를 사용한다. to_sql 보다 더 많은 오버헤드를 갖고 더 많은 권한이 요구되므로, 1000개 이상의 row를 갖는 경우에만 사용하는것을 권장한다고 되어 있다.

💡 Load Pandas DataFrame as a Table on Amazon Redshift using parquet files on S3 as stage.
This is a HIGH latency and HIGH throughput alternative to wr.redshift.to_sql() to load large DataFrames into Amazon Redshift through the SQL COPY command
This strategy has more overhead and requires more IAM privileges than the regular wr.redshift.to_sql() function, so it is only recommended to inserting +1K rows at once.

wr.redshift.copy(
    df=df,  #적재하고자 하는 판다스 데이터프레임
    # stage parquet파일을 저장할 위치로, 해당파일에 데이터를 쓰고 지우는것이 자동으로 이루어지기 때문에 empty 상태여야만 함!
    path=f"s3://{self.bucket_name}/dw_stage/",
    con=wr_con,  # 앞에서 생성한 커넥션
    schema=redshift_schema,  #레드시프트 테이블 스키마
    table=table_name,  # 레드시프트 테이블 이름
    mode=copy_mode,  # ["append", "overwrite", "upsert"] default: "append"
    overwrite_method=overwrite_method,  # ["drop", "cascade", "truncate", "delete"] default: "drop"
)

Parquet파일을 바로 가져다가 테이블에 Load

path에 해당하는 값이 stage 파일 위치가 아닌, parquet파일이 저장되어 있는 위치라는것만 copy 메소드와 다르다.

wr.redshift.copy_from_files(
    path=f"s3://{s3_bucket}/{s3_copy_key}",
    con=wr_con,
    schema=table_schema,
    table=table_name,
    mode=copy_mode
)

개발새발