배경

GCS의 csv 데이터를 S3로 transfer하고 싶습니다.
GCS-S3 전송에는 GoogleCloudStorageToS3Operator을 사용한다.
S3의 비밀 정보 관리는 Airflow 연결에서 관리한다.

Airflow 연결 준비

S3에 대한 비밀 정보는 Airflow 연결을 사용하여 관리합니다.
Airflow 콘솔 화면에서 Airflow 연결 정보를 등록합니다.
새로운 Airflow 연결 만들기

↓등록 내용



Login: AWS 액세스 키
Password: AWS 액세스 비밀 키

샘플 DAG

import airflow
from datetime import timedelta
from airflow import DAG
from airflow.contrib.operators.gcs_to_s3 import GoogleCloudStorageToS3Operator

default_args = {
    "owner": "airflow",
    "depends_on_past": False,
    "start_date": airflow.utils.dates.days_ago(1),
    "execution_timeout": timedelta(minutes=30),
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "transfer_gcs_to_s3",
    default_args=default_args,
    catchup=False,
    schedule_interval="0 0 * * *",
)

GoogleCloudStorageToS3Operator(
    task_id="transfer_gcs_to_s3",
    bucket='gcs_bucket_name',
    prefix="gcs/path/to/file.csv",
    dest_aws_conn_id='gcs_to_s3_sample', # Airflow接続に登録したid
    dest_s3_key="s3://s3-bucket-name/",
    replace=True,
    dag=dag
)

참고

사용법 샘플 git: [AIRFLOW-6147] [AIP-21] Rename GoogleCloudStorageToS3Operator

S3 시크릿을 Airflow 연결로 처리하는 방법: Airflow s3 connection using UI