Cloud Composer 1은 사후 유지보수 모드 상태입니다. Google은 Airflow의 새 버전, 버그 수정, 보안 업데이트를 비롯한 Cloud Composer 1 추가 업데이트를 출시하지 않습니다. Cloud Composer 2로 마이그레이션하는 것이 좋습니다.

Dataplex를 사용한 데이터 계보

Cloud Composer 1 | Cloud Composer 2 | Cloud Composer 3

데이터 계보 통합 정보

데이터 계보는 시스템을 통해 데이터가 이동하는 방식, 즉 데이터의 출처, 데이터가 전달되는 위치, 데이터에 적용되는 변환을 추적할 수 있는 Dataplex 기능입니다. 데이터 계보는 다음에서 사용할 수 있습니다.

Airflow 버전 2.2.5 이상을 실행하는 Cloud Composer2 버전 2.1.2 이상의 환경
데이터 계보를 지원하는 Data Catalog 리전과 동일한 리전의 Cloud Composer 2 환경

Cloud Composer 환경에서 이 기능을 사용 설정하면 지원되는 연산자를 활용하는 DAG를 실행할 때 Cloud Composer가 계보 정보를 Data Lineage API에 보고합니다.

그런 다음 다음을 사용하여 해당 정보에 액세스할 수 있습니다.

Data Lineage API
Dataplex에서 지원되는 Data Catalog 항목의 계보 시각화 그래프. Dataplex 문서의 계보 시각화 그래프를 참조하세요.

지원되는 연산자

다음 연산자는 Cloud Composer에서 자동 계보 보고를 지원합니다.

airflow.providers.google.cloud.operators.bigquery.BigQueryExecuteQueryOperator
airflow.providers.google.cloud.operators.bigquery.BigQueryInsertJobOperator
airflow.providers.google.cloud.transfers.bigquery_to_bigquery.BigQueryToBigQueryOperator
airflow.contrib.operators.bigquery_to_gcs.BigQueryToCloudStorageOperator
airflow.providers.google.cloud.transfers.bigquery_to_gcs.BigQueryToGCSOperator
airflow.providers.google.cloud.transfers.gcs_to_bigquery.GCSToBigQueryOperator
airflow.contrib.operators.gcs_to_bq.GoogleCloudStorageToBigQueryOperator
airflow.providers.google.cloud.operators.dataproc.DataprocSubmitJobOperator

예를 들어 다음 태스크를 실행하는 경우:

task = BigQueryInsertJobOperator(
    task_id='snapshot_task',
    dag=dag,
    location='<dataset-location>',
    configuration={
        'query': {
            'query': 'SELECT * FROM dataset.tableA',
            'useLegacySql': False,
            'destinationTable': {
                'project_id': GCP_PROJECT,
                'dataset_id': 'dataset',
                'table_id': 'tableB',
            },
        }
    },
)

Dataplex UI에 다음 계보 그래프가 생성됩니다.

Dataplex UI에서 계보 그래프 예시 — **그림 1.** Dataplex UI에서 BigQuery 테이블의 샘플 계보 그래프

Cloud Composer 기능 고려사항

데이터 계보를 보고하는 각 Airflow 태스크 실행은 다음을 수행합니다.

계보 프로세스에 대한 하나의 RPC 생성 또는 업데이트 요청
계보 실행에 대한 하나의 RPC 생성 또는 업데이트 요청
계보 이벤트를 만드는 하나 이상의 RPC 요청(대부분의 경우 0 또는 1)

이러한 항목에 대한 자세한 내용은 Dataplex 문서의 계보 정보 모델 및 Lineage API 참조를 확인하세요.

내보낸 계보 트래픽에는 Data Lineage API의 할당량이 적용됩니다. Cloud Composer는 쓰기 할당량을 사용합니다.

계보 데이터 처리와 관련된 가격 책정에는 계보 가격 책정이 적용됩니다. 데이터 계보 고려사항을 참조하세요.

성능 영향

데이터 계보는 Airflow 태스크 실행이 끝날 때 보고됩니다. 데이터 계보 보고에는 평균적으로 약 1~2초가 걸립니다.

이는 태스크 자체의 성능에 영향을 미치지 않습니다. Lineage API에 계보가 성공적으로 보고되지 않으면 Airflow 태스크가 실패하지 않습니다. 기본 연산자 로직에는 영향을 미치지 않지만 계보 데이터 보고를 고려하여 전체 태스크 인스턴스는 좀 더 오래 실행됩니다.

데이터 계보를 보고하는 환경에서는 데이터 계보를 보고하는 데 필요한 추가 시간이 발생하므로 관련 비용이 약간 증가합니다.

규정 준수

데이터 계보는 VPC 서비스 제어와 같은 기능을 다양한 지원 수준으로 제공합니다. 데이터 계보 고려사항을 검토하여 지원 수준이 환경 요구사항과 일치하는지 확인하세요.

데이터 계보 통합 지원

Cloud Composer의 데이터 계보 통합은 환경별로 관리됩니다. 즉, 이 기능을 사용 설정하려면 다음 두 단계를 수행해야 합니다.

프로젝트에서 Data Lineage API를 사용 설정합니다.
특정 Cloud Composer 환경에서 데이터 계보 통합을 사용 설정합니다.

시작하기 전에

환경을 만들 때 다음 조건이 충족되면 데이터 계보 통합이 자동으로 사용 설정됩니다.

프로젝트에 Data Lineage API가 사용 설정되어 있습니다. 자세한 내용은 Dataplex 문서의 Data Lineage API 사용 설정을 참조하세요.

주의: Data Lineage API를 사용 설정하면 Dataplex에서 BigQuery, Cloud Data Fusion, Dataproc에 대한 데이터를 자동으로 수집합니다. 이는 데이터 계보 통합으로 Cloud Composer 환경을 만들지 않은 경우에도 발생합니다.
Airflow에는 커스텀 계보 백엔드가 구성되지 않습니다.

기존 환경의 경우 언제든지 데이터 계보 통합을 사용 설정하거나 중지할 수 있습니다.

필요한 역할

데이터 계보와 통합하려면 Cloud Composer 환경 서비스 계정에 다음 권한을 추가해야 합니다.

기본 서비스 계정의 경우 변경할 필요가 없습니다. 기본 서비스 계정에는 필요한 권한이 포함되어 있습니다.
사용자 관리 서비스 계정의 경우: 서비스 계정에 Composer 작업자(roles/composer.worker) 역할을 부여합니다. 이 역할에는 필요한 모든 데이터 계보 권한이 포함됩니다.

자세한 내용은 Dataplex 문서의 계보 역할 및 권한을 참조하세요.

Cloud Composer에서 데이터 계보 사용 설정

콘솔

Google Cloud 콘솔에서 환경 페이지로 이동합니다.

환경으로 이동
환경 목록에서 환경 이름을 클릭합니다. 환경 세부정보 페이지가 열립니다.
환경 구성 탭을 선택합니다.
Dataplex 데이터 계보 통합 섹션에서 수정을 클릭합니다.
Dataplex 데이터 계보 통합 패널에서 Dataplex 데이터 계보와의 통합 사용 설정을 선택하고 저장을 클릭합니다.

gcloud

--enable-cloud-data-lineage-integration 인수를 사용합니다.

gcloud composer environments update ENVIRONMENT_NAME \
    --location LOCATION \
    --enable-cloud-data-lineage-integration

다음과 같이 바꿉니다.

ENVIRONMENT_NAME을 환경 이름으로 바꿉니다.

이름은 소문자로 시작해야 합니다. 이어서 최대 62자(영문 기준)의 소문자, 숫자 또는 하이픈이 와야 하며 하이픈으로 끝나서는 안 됩니다 환경 이름은 환경의 하위 구성요소를 만드는 데 사용되므로 Cloud Storage 버킷 이름으로도 유효한 이름으로 지정해야 합니다. 제한사항 목록은 버킷 이름 지정 가이드라인을 참조하세요.
LOCATION을 환경의 리전으로 바꿉니다.

위치는 환경의 GKE 클러스터가 있는 리전입니다.

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

gcloud composer environments update example-environment \
    --location us-central1 \
    --enable-cloud-data-lineage-integration

커스텀 계보 이벤트 전송

자동 계보 보고에서 지원되지 않는 연산자의 계보를 보고하려면 커스텀 계보 이벤트를 전송할 수 있습니다.

예를 들어 다음과 같이 커스텀 이벤트를 전송할 수 있습니다.

BashOperator: 태스크 정의에서 inlets 또는 outlets 파라미터를 수정합니다.
PythonOperator: 태스크 정의에서 task.inlets 또는 task.outlets 파라미터를 수정합니다. inlets 매개변수에 AUTO를 사용하면 업스트림 태스크의 outlets와 동일하게 값을 설정합니다.

예를 들어 다음 태스크를 실행한 경우


from airflow.composer.data_lineage.entities import BigQueryTable
from airflow.lineage import AUTO

…

bash_task = BashOperator(
   task_id='bash_task',
   dag=dag,
   bash_command='sleep 0',
   inlets=[BigQueryTable(
       project_id=GCP_PROJECT,
       dataset_id='dataset',
       table_id='table1',
   )],
   outlets=[BigQueryTable(
       project_id=GCP_PROJECT,
       dataset_id='dataset',
       table_id='table2',
   )]
)


def _python_task(task):
   task.inlets.append(BigQueryTable(
       project_id=GCP_PROJECT,
       dataset_id='dataset',
       table_id='table3',
   ))

   task.outlets.append(BigQueryTable(
       project_id=GCP_PROJECT,
       dataset_id='dataset',
       table_id='table4',
   ))


python_task = PythonOperator(
   task_id='python_task',
   dag=dag,
   python_callable=_python_task,
   inlets=[AUTO],
)

bash_task >> python_task

Dataplex UI에 다음 계보 그래프가 생성됩니다.

Dataplex UI에서 커스텀 이벤트에 대한 계보 그래프 예시 — **그림 2.** Dataplex UI에서 여러 BigQuery 테이블의 샘플 계보 그래프

Cloud Composer에서 데이터 계보 중지

Cloud Composer 환경에서 계보 통합을 중지해도 Data Lineage API는 중지되지 않습니다. 프로젝트의 계보 보고를 완전히 사용 중지하려면 Data Lineage API도 사용 중지하세요. 서비스 사용 중지를 참조하세요.

콘솔

Google Cloud 콘솔에서 환경 페이지로 이동합니다.

환경으로 이동
환경 목록에서 환경 이름을 클릭합니다. 환경 세부정보 페이지가 열립니다.
환경 구성 탭을 선택합니다.
Dataplex 데이터 계보 통합 섹션에서 수정을 클릭합니다.
Dataplex 데이터 계보 통합 패널에서 Dataplex 데이터 계보와의 통합 사용 중지를 선택하고 저장을 클릭합니다.

gcloud

--disable-cloud-data-lineage-integration 인수를 사용합니다.

gcloud composer environments update ENVIRONMENT_NAME \
    --location LOCATION \
    --disable-cloud-data-lineage-integration

다음과 같이 바꿉니다.

ENVIRONMENT_NAME을 환경 이름으로 바꿉니다.

이름은 소문자로 시작해야 합니다. 이어서 최대 62자(영문 기준)의 소문자, 숫자 또는 하이픈이 와야 하며 하이픈으로 끝나서는 안 됩니다 환경 이름은 환경의 하위 구성요소를 만드는 데 사용되므로 Cloud Storage 버킷 이름으로도 유효한 이름으로 지정해야 합니다. 제한사항 목록은 버킷 이름 지정 가이드라인을 참조하세요.
LOCATION을 환경의 리전으로 바꿉니다.

위치는 환경의 GKE 클러스터가 있는 리전입니다.

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

gcloud composer environments update example-environment \
    --location us-central1 \
    --disable-cloud-data-lineage-integration

Cloud Composer에서 계보 로그 보기

Dataplex 데이터 계보 통합 섹션의 환경 구성 페이지에 있는 링크를 사용하여 데이터 계보와 관련된 로그를 검사할 수 있습니다.

문제 해결

계보 데이터가 Lineage API에 보고되지 않거나 Dataplex에서 표시되지 않으면 다음 문제 해결 단계를 시도해 보세요.

Cloud Composer 환경의 프로젝트에서 Data Lineage API가 사용 설정되어 있는지 확인합니다.
Cloud Composer 환경에서 데이터 계보 통합이 사용 설정되어 있는지 확인합니다.
사용하는 연산자가 자동 계보 보고 지원에 포함되어 있는지 확인합니다. 지원되는 Airflow 연산자를 참조하세요.
Cloud Composer의 계보 로그에서 문제가 있는지 확인하세요.