AWS DataZone에서 OpenLineage 기반의 Airflow 데이터 계보 그리기

Amazon Web ServicesAWS DataZone에서 OpenLineage 기반의 Airflow 데이터 계보 그리기 | Amazon Web Services

AWS DataZone에서 OpenLineage 기반의 Airflow 데이터 계보 그리기 | Amazon Web Services

배경 Airflow는 데이터 마트(Data Mart)를 포함한 데이터 파이프라인 구축 및 관리에서 매우 널리 사용되는 도구입니다. 이러한 Airflow에서 데이터 계보가 중요한 이유는 데이터의 출처와 변환 과정을 명확히 추적할 수 있어 데이터의 신뢰성을 보장하고, 문제 발생 시 원인을 빠르게 파악할 수 있기 때문입니다. 또한, 데이터 계보는 규제 준수와 감사 요구사항을 충족시키는 데 도움을 주며, 데이터 파이프라인의 변경이 […]

Amazon Web Services

배경

• Airflow는 Data mart를 포함한 데이터 파이프라인 구축 및 관리에서 널리 사용되는 도구

데이터 계보를 통해 복잡한 데이터 파이프라인을 효율적으로 운영하며, 데이터 기반 의사결정의 신뢰성과 조직의 데이터 활용 능력을 향상

 
• 데이터 출처와 변환 과정을 명확히 추적할 수 있어 데이터의 신뢰성을 보장 & 문제 발생 시 원인 빠르게 파악
• 데이터 파이프라인의 변경이 미치는 영향을 사전에 분석 → 위험 최소화
+) 데이터 변경하는 것보다 차라리 컬럼 하나 새로 만드는 방향으로 업무 진행하는 흐름
→ 데이터 품질 보장, 다운타임을 줄임, 시스템 안정성을 높임
 
• 데이터 계보는 규제 준수와 감사 요구사항 파악에 도움
+) 금융권에서 계속 감사 들어오는 거 생각해보면 규제 준수는 점점 강화되고 감사 요구사항은 많아지는 방향으로만 갔다
→ 조직 내 데이터 거버넌스 정책을 강화하여 데이터 소유권, 사용 권한, 보안 정책을 효과적으로 관리

솔루션 개요

• Amazon DataZone의 데이터 계보 기능을 활용하여 Airflow의 데이터 계보를 그릴 수 있음
• Amazon DataZone 서비스는 DataZone 라이프 사이클 내에서 발생한 게시와 구독 이벤트에 대해서만 데이터 계보가 표현됨 → 그 외에는 직접 작성해야 함
• AWS Lambda를 사용하여 Airflow에 대한 계보를 OpenLineage 표준에 맞게 DataZone에 업데이트 하는 방법을 설명

솔루션 아키텍쳐

• Production 계정과 Governance 계정은 Transit Gateway 를 통해 연결되어 네트워크 통신이 가능함

• Production 계정에서 데이터의 저장 및 메타데이터 생성 작업이 이뤄짐

1. RDS는 데이터를 저장하고, 사용자의 요청에 따라 View Table을 생성

2. AWS Glue Crawler가 RDS 데이터를 스캔하여 메타 데이터를 자동으로 생성, Glue Data Catalog에 저장
 

3. MWAA(Amazon Managed Workflows for Apache Airflow)에서 매일 View Table을 갱신하는 Airflow Job이 배치 형태로 동작

4. 동작할 때마다 로그 정보를 CloudWatch에 저장 → 이 정보를 Assume Role을 이용해서 Governance 계정에서 조회

5. Governance 계정에서는 Airflow에 대한 데이터 계보 작성과 사용자 접근 제어가 이뤄짐
 

6. Lambda 함수는 매시간 Airflow Job에 대한 로그를 가져옴

7. 로그에서 프로시저 함수를 가져오고 함수 이름을 이용해 프로시저 정의 요청 쿼리문을 이용해 Input TB, Output TB 이름을 조회
+) Jennifer에서 수행문 중에 SQL 쿼리 확인할 수 있음. 이처럼 쌓이는 SQL 쿼리 로그를 분석한다는 의미

import os
import boto3
import json
import re
from openlineage_sql import parse
from psycopg2.sql import SQL, Identifier, Literal

procedure_name = re.search(r'\b([a-zA-Z_]+)\b(?=\()', query).group(1)

# 프로시저에 대한 정의를 요청하는 쿼리 요청
sql_query = """
    SELECT PROSRC
    FROM PG_CATALOG.PG_PROC
    WHERE PRONAME = {};
"""
procedure_definition = run_query(config.db_secret_name, 
    SQL(sql_query).format(
        Literal(procedure_name)
    )             
)

if not procedure_definition:
    print(f"[ERROR] {procedure_name} 프로시저에 대한 정의가 존재하지 않습니다.")
    return 'error'
else:
    dml_pattern = r'(DELETE.*?;|INSERT.*?;|UPDATE.*?;|MERGE.*?;)'
    dml_statements = re.findall(dml_pattern, procedure_definition[0]['prosrc'], re.DOTALL | re.IGNORECASE)

    sql_query = ''.join(dml_statements)

    # openlineage_sql을 이용해 파싱한다.
    meta = parse([sql_query])

    in_tables = meta.in_tables
    out_table = meta.out_tables[0]

8. 조회한 정보과 Airflow Job 정보를 조합한 후, 데이터 계보를 Amazon DataZone에 업데이트

결론