Airflow 연결 관리

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

이 페이지에서는 사용자 환경에서 Airflow 연결을 관리하고 DAG에서 액세스하는 방법을 설명합니다.

Airflow 연결 정보

Airflow 연결은 사용자 인증 정보 및 사용자 이름, 연결 문자열, 비밀번호와 같은 기타 연결 정보를 저장합니다. DAG는 연결을 사용하여 DAG에서 Google Cloud 및 다른 서비스의 리소스를 통신하고 액세스합니다.

DAG의 Airflow 연산자는 연산자의 기본 연결을 사용하거나 커스텀 연결 이름을 지정합니다.

연결 보안 정보

대부분의 Airflow 연산자는 사용자 인증 정보를 직접 허용하지 않습니다. 대신 Airflow 연결을 사용합니다.

새 환경을 만들면 Cloud Composer가 환경의 고유 영구 퍼넷 키를 생성하고 기본적으로 연결 엑스트라를 보호합니다. Airflow UI의 구성 페이지에서 fernet_key를 확인할 수 있습니다.

Airflow에서 연결 및 비밀번호가 보호되는 방법에 대한 자세한 내용은 연결 보안 및 민감한 정보 마스킹을 참조하세요.

연결 유형 정보

Airflow는 다양한 유형의 연결을 사용하여 특정 서비스에 연결합니다. 예를 들어 Google Cloud 연결 유형은 Google Cloud의 다른 서비스에 연결됩니다. 또 다른 예시로 S3 연결 유형은 Amazon S3 버킷에 연결합니다.

Airflow에 연결 유형을 추가하려면 해당 연결 유형으로 PyPI 패키지를 설치합니다. 일부 패키지는 사용자 환경에 사전 설치되어 있습니다. 예를 들어 커스텀 PyPI 패키지를 설치하지 않고도 apache-airflow-providers-google 패키지의 연결을 사용할 수 있습니다.

사전 구성된 연결

Cloud Composer는 환경에서 다음과 같은 기본 연결을 구성합니다. 이러한 연결을 사용하여 프로젝트를 구성하지 않고도 프로젝트의 리소스에 액세스할 수 있습니다.

• google_cloud_default
• bigquery_default
• google_cloud_datastore_default
• google_cloud_storage_default

Secret Manager에서 연결 추가

Airflow에 추가하지 않고 Secret Manager에 연결을 저장할 수 있습니다. 사용자 인증 정보 및 기타 민감한 정보를 저장할 때는 이 방법을 사용하는 것이 좋습니다.

Secret Manager에서 연결을 추가하려면 다음 안내를 따르세요.

1. 환경에 Secret Manager를 구성합니다.

2. 연결 패턴과 일치하는 이름으로 보안 비밀을 추가합니다.

   예를 들면 airflow-connections-example_connection입니다. DAG에서 프리픽스(example_connection) 없이 연결 이름을 사용합니다.

3. 연결에 매개변수를 추가합니다.

   JSON 형식

   연결의 JSON 표현을 보안 비밀 값으로 추가합니다. 예를 들면 다음과 같습니다.

   {
     "conn_type": "mysql",
     "host": "example.com",
     "login": "login",
     "password": "password",
     "port": "9000"
   }
   

   JSON 연결 형식에 대한 자세한 내용은 Airflow 문서를 참조하세요.

   URI 형식

   연결의 URI 표현을 보안 비밀 값으로 추가합니다.

   • 보안 비밀은 연결의 URI 표현을 저장해야 합니다. 예를 들면 mysql://login:password@example.com:9000입니다.

   • URI는 URL로 인코딩되어야 합니다. 예를 들어 공백 기호가 있는 비밀번호는 다음과 같이 URL로 인코딩되어야 합니다. mysql://login:secret%20password@example.com:9000.

   Airflow에는 연결 URI 생성을 위한 편의 메서드가 있습니다. JSON 엑스트라를 사용하여 복잡한 URL을 인코딩하는 방법에 대한 예시는 Airflow 문서를 참조하세요.

4. Secret Manager에서 모든 연결 매개변수를 올바르게 읽는지 확인합니다.

Airflow에서 연결 추가

Secret Manager에 연결을 저장하는 대신 Airflow에 연결을 저장할 수 있습니다.

Airflow에서 연결을 추가하려면 다음 안내를 따르세요.

gcloud composer environments run ENVIRONMENT_NAME \
  --location LOCATION \
  connections add -- \
    --conn-type "mysql" \
    --conn-host "example.com" \
    --conn-port "9000" \
    --conn-login "login" \
    --conn-password "password" \
    example_connection

gcloud composer environments run ENVIRONMENT_NAME \
  --location LOCATION \
  connections add -- \
    --conn-uri "mysql://login:password@example.com:9000" \
    example_connection

Airflow가 연결을 올바르게 읽는지 확인

Google Cloud CLI를 통해 connections get Airflow CLI 명령어를 실행하여 연결이 올바르게 읽히는지 확인할 수 있습니다. 예를 들어 Secret Manager에 연결을 저장하면 보안 비밀에서 연결의 모든 매개변수를 읽을 수 있는지 확인하는 방법이 제공됩니다.

gcloud composer environments run ENVIRONMENT_NAME \
    --location LOCATION \
    connections get \
    -- CONNECTION_NAME

다음과 같이 바꿉니다.

• ENVIRONMENT_NAME: 환경 이름
• LOCATION: 환경이 위치한 리전
• CONNECTION_NAME을 연결 이름으로 바꿉니다. 연결이 Secret Manager에 저장된 경우 연결 프리픽스 없이 연결 이름을 사용합니다. 예를 들어 airflow-connections-example_connection_json 대신 example_connection을 지정합니다.

예를 들면 다음과 같습니다.

gcloud composer environments run example-environment \
    --location us-central1 \
    connections get \
    -- example_connection -o json

DAG에서 Airflow 연결 사용

이 섹션에서는 DAG에서 연결에 액세스하는 방법을 설명합니다.

Secret Manager 연결 사용

프리픽스 없이 연결 이름을 사용합니다. 예를 들어 보안 비밀 이름이 airflow-connections-aws_s3이면 aws_s3을 지정합니다.

transfer_dir_from_s3 = S3ToGCSOperator(
    task_id='transfer_dir_from_s3',
    aws_conn_id='aws_s3',
    prefix='data-for-gcs',
    bucket='example-s3-bucket-transfer-operators',
    dest_gcs='gs://us-central1-example-environ-361f4221-bucket/data/from-s3/')

Secret Manager에 기본 연결을 저장하는 경우 연결 이름을 생략할 수 있습니다. 연산자에서 사용하는 기본 연결 이름을 가져오려면 특정 연산자에 대한 Airflow 문서를 참조하세요. 예를 들어 S3ToGCSOperator Airflow 연산자는 기본적으로 aws_default 연결을 사용합니다. 이 기본 연결을 airflow-connections-aws_default라는 보안 비밀에 저장할 수 있습니다.

Airflow에 저장된 연결 사용

Airflow에 정의된 연결 이름을 사용합니다.

transfer_dir_from_s3 = S3ToGCSOperator(
    task_id='transfer_dir_from_s3',
    aws_conn_id='aws_s3',
    prefix='data-for-gcs',
    bucket='example-s3-bucket-transfer-operators',
    dest_gcs='gs://us-central1-example-environ-361f4221-bucket/data/from-s3/')

연산자에 기본 연결을 사용하려면 연결 이름을 생략합니다. 연산자에서 사용하는 기본 연결 이름을 가져오려면 특정 연산자에 대한 Airflow 문서를 참조하세요. 예를 들어 S3ToGCSOperator Airflow 연산자는 기본적으로 aws_default 연결을 사용합니다.

문제 해결

환경에서 Secret Manager에 저장된 보안 비밀에 액세스할 수 없는 경우:

1. 해당 환경에 Secret Manager가 구성되었는지 확인합니다.

2. Secret Manager의 연결 이름이 Airflow에서 사용하는 연결에 해당하는지 확인합니다. 예를 들어 이름이 example_connection인 연결의 경우 보안 비밀 이름은 airflow-connections-example_connection입니다.

3. Airflow가 연결을 올바르게 읽는지 확인합니다.

다음 단계

• Secret Manager 구성
• 이메일 알림 구성
• DAG 작성