[시행착오] 데브코스 3차 프로젝트(Airflow 활용 ETL)

프로젝트에서 발생했던 오류 정리

이번 데브코스에서 프로젝트를 진행하면서 발생했던 오류와 해결 과정을 정리해보려고 한다. 나는 Airflow의 DAG를 작성하는 역할을 맡았기에 데이터를 S3, Redshift에 저장하는 과정에서 에러를 마주치게 되었다.

S3에 저장한 csv 열었을 때 한글 깨짐

문제 발생

API를 통해 데이터를 가져와 데이터 프레임 형태로 만들고, S3에 csv로 저장해 주었다. 에러가 발생하지 않아서 올바르게 작동하나 싶었지만, aws 웹 콘솔로 접속해서 S3의 파일을 직접 열었을 때 한글이 깨지는 것을 확인할 수 있었다. 영어와 숫자는 제대로 나오기 때문에 인코딩-디코딩 과정에서 오류가 있을 것이라 생각했다.

문제 해결

결론적으로 구글링을 많이 해봤지만, S3의 파일을 다운로드하고 열었을 때 한글이 깨지는 오류 자체는 해결할 수 없었다. 그러나 Python에서 파일을 가져와 decode 후 확인해 보면, 한글이 깨지지 않았다. S3에서 직접 csv를 다운로드하여 열면 디코딩이 제대로 되지 않는다고 추측하였다. 결국 데이터는 python을 통해서만 확인하였다.

• 아래의 코드로 확인 시 한글이 깨지지 않음

# S3 버킷과 파일 경로
bucket_name = 'dev-3-2-bucket'
key = 'locgoRegnVisitrDDList.csv'

# S3 객체에서 CSV 파일을 가져오기
response = s3_client.get_object(Bucket=bucket_name, Key=key)
data = response['Body'].read().decode('utf-8')

# CSV 데이터를 데이터프레임으로 변환
df = pd.read_csv(StringIO(data), encoding='utf-8')

Redshift에 COPY 시 에러 발생

문제 발생

Airflow의 S3ToRedshiftOperator를 사용해 Redshift로 COPY 하던 과정에서 정상 실행이 되지 않고 에러가 발생하였다. Redshift의 stl_load_errors 테이블을 확인하여 에러를 확인해 보았는데, 두 가지 오류가 있었다. 추가로 COPY를 진행하기 위해 S3 Connection을 진행하는 과정에서 AccessDenied가 발생하였다.

1. '41234.00'의 문자열을 INT로 지정하고, COPY를 진행하자 '.' 문자가 있다고 에러 발생
2. 날짜 형식이 '20240101'이지만, DATE 기본 형식이 '2024-01-01'이라서 에러 발생
3. IAM 사용자 권한 부족으로 S3 Connection 에러 발생 (S3 AccessDenied)

문제 해결 (1)

해당 컬럼의 데이터 타입을 float -> int 순으로 변환해서 S3에 저장하여 해결하였다.

df['touNum'] = df['touNum'].astype(float)
df['touNum'] = df['touNum'].astype(int)

문제 해결 (2)

DATE의 기본 형태는 '2024-01-01'이다. 그러나 저장된 데이터의 형태는 '20240101'로 대시(-)가 없는 형태이다. S3ToRedshiftOperator의 copy_options에 날짜 형식을 지정할 수 있는 옵션을 활용하여 문제를 해결하였다.

• DATEFORMAT 'YYYYMMDD' : DATE 형식을 '20240101' 형태로 지정
• FILLRECORD : 모든 레코드에 적용

load_to_redshift = S3ToRedshiftOperator(
            task_id='load_to_redshift',
            s3_bucket=s3_bucket,
            s3_key=s3_key,
            schema=schema,
            table=table,
            copy_options=[
                'csv',
                'IGNOREHEADER 1',
                "DATEFORMAT 'YYYYMMDD'",
                'FILLRECORD'
            ],
            method='REPLACE',
            aws_conn_id='dev-3-2-bucket',
            redshift_conn_id='redshift_dev_db'
)

문제 해결 (3)

COPY 시 에러가 발생할 때의 IAM 사용자의 S3 권한은 GetObject, PutObject, DeleteObject였지만, AccessDenied가 발생하였다. 여기에 ListBucket 권한을 추가하자 에러 없이 코드가 작동하였다.

"Action": [
    "s3:GetObject",
    "s3:PutObject",
    "s3:DeleteObject",
    "s3:ListBucket"
],

웹 콘솔로 S3 접근 가능한 IAM 생성

문제 발생

DAG를 작성하던 사람이 한 분 더 있었는데, S3 버킷 관리자는 나였다. 그래서 다른 분은 웹 콘솔을 통해 접속을 할 수 없는 상태였다. 그러나 작업을 위해 필요할 수도 있다 생각해서 IAM 사용자를 추가로 생성하고자 하였다. 이 과정에서 최소한의 권한을 가진 정책으로 생성하려고 했지만, S3에 접근이 아예 불가하게 생성되었다.

문제 해결

일단 즉시 사용할 수 있도록 AmazonS3FullAccess 정책을 적용하자 접속이 가능한 것을 확인하였다. 그러나 보안 이슈를 방지하기 위해 최소한의 권한으로 제공하기 위한 정책을 찾아보면 좋을 것 같다.

S3Hook로 csv를 가져올 때 에러 발생

문제 발생

S3에서 읽은 데이터를 decode 하여 df에 저장하는 코드인데, 이 과정에서 에러가 발생하였다. 그러나 Airflow가 아닌 boto3을 활용해서 똑같이 진행했을 때는 제대로 수행했던 코드였기에 의아했다.

print("Get csv file from S3")
response = hook.read_key(key, bucket_name)
data = response.decode('utf-8')
df = pd.read_csv(StringIO(data))
print("Done")

문제 해결

이전과 달라진 것은 read_key였기에 해당 코드를 Airflow 공식 문서에서 살펴보았다. read_key의 반환값을 보면 객체를 읽은 뒤 decode 하는 과정까지 포함되어 있다. 그러나 df에 저장하기 전 decode를 또 진행하기에 에러가 발생한 것이다. read_key 이후 decode 코드를 삭제하니 올바르게 작동하였다.

• 공식 문서 코드

def read_key(self, key, bucket_name=None):
    """
    Reads a key from S3
    :param key: S3 key that will point to the file
    :type key: str
    :param bucket_name: Name of the bucket in which the file is stored
    :type bucket_name: str
    """

    obj = self.get_key(key, bucket_name)
    return obj.get()['Body'].read().decode('utf-8')

• 수정한 코드

print("Get csv file from S3")
response = hook.read_key(key, bucket_name)
df = pd.read_csv(StringIO(response))
print("Done")