datahub-airflow-plugin-lite

Airflow 실행 메타데이터와 lineage를 DataHub로 보내는 런타임 의존성 0개 플러그인입니다. 표준 라이브러리와 이미 설치된 Airflow만 사용하므로 MWAA 등 Airflow constraints 파일과 충돌하지 않습니다.

설치

pip install datahub-airflow-plugin-lite-0.1.1-py3-none-any.whl

공식 플러그인(acryl-datahub-airflow-plugin)과 동시에 설치하지 마세요. 두 리스너가 같은 엔티티를 중복 전송합니다.

설정

airflow.cfg의 [datahub] 섹션(또는 AIRFLOW__DATAHUB__* 환경변수)을 읽습니다. 키 이름은 공식 플러그인과 같습니다.

[datahub]
# GMS 주소/토큰을 직접 지정하거나, 생략하면 conn_id 커넥션(host=GMS URL, password=토큰)을 사용
gms_url = https://datahub-gms.example.com
token = <personal access token>
conn_id = datahub_rest_default
cluster = prod
platform_instance =
capture_ownership_info = true
capture_ownership_as_group = false
capture_tags_info = true
capture_executions = true
materialize_iolets = true
capture_airflow_assets = true
enable_datajob_lineage = true
enable_openlineage_facets = true
openlineage_timeout_sec = 5
supplement_with_sql_parser = true
athena_include_result_location = false
dag_filter_str = {"allow": [".*"], "deny": []}
timeout_sec = 10
retries = 2
ssl_verify = true

운영 스위치도 공식 플러그인과 같습니다.

• AIRFLOW_VAR_DATAHUB_AIRFLOW_PLUGIN_DISABLE_LISTENER=true: 리스너 즉시 중지 (kill switch)
• DATAHUB_AIRFLOW_PLUGIN_RUN_IN_THREAD, DATAHUB_AIRFLOW_PLUGIN_RUN_IN_THREAD_TIMEOUT: 백그라운드 스레드 실행 여부와 대기 시간

동작

전송은 GMS의 POST /aspects?action=ingestProposal 하나만 사용합니다. 모든 처리는 데몬 스레드에서 실행되고, 예외는 로그로만 남기므로 태스크 실패로 이어지지 않습니다.

Lineage 소스

1. inlets/outlets의 Dataset/Urn 객체 또는 urn:li:... 문자열
   • datahub_airflow_plugin_lite.entities.Dataset/Urn
   • 공식 플러그인의 datahub_airflow_plugin.entities 객체도 urn 속성이 있어 그대로 인식합니다.
2. Airflow 네이티브 Dataset(2.x)/Asset(3.x). URI 스킴으로 플랫폼을 판별합니다.
3. OpenLineage provider (기본 소스): operator의 get_openlineage_facets_on_complete/start() 결과를 사용합니다. 몽키패치는 하지 않습니다.
   • 정규화: provider 버전마다 다른 결과를 DataHub ingestion URN으로 맞춥니다(ol_normalize.py).
     • Athena: AwsDataCatalog. 접두어를 제거하고 소문자로 바꿉니다.
     • S3: 버킷과 키를 합쳐 make_s3_urn과 같은 규칙을 적용합니다.
     • Snowflake: 소문자로 바꿉니다.
   • 버전 의존 값 제거: 버전마다 다르게 나오는 Athena 쿼리 결과 위치는 제외합니다.
   • 시간 제한: OpenLineage 호출은 openlineage_timeout_sec(기본 5초)까지만 기다립니다.
4. 내장 SQL 파서 (보완): OpenLineage가 버전에 따라 놓친 테이블을 채웁니다. 예: provider 2.x의 CTE 입력 누락, amazon 8.x의 Athena 테이블 누락. 이 단계는 OpenLineage 결과를 지우지 않습니다.
   • DB에 접속하지 않고 SQL 문과 Connection 정보(conn_type, 기본 DB/스키마)만 사용합니다.

lineage 수집이 실패해서 결과가 불완전할 수 있으면 dataJobInputOutput을 보내지 않습니다. 이미 저장된 lineage를 덮어쓰지 않기 위해서입니다. 예: OpenLineage가 실패했고 SQL도 분석할 수 없는 경우. DataJob 정보와 실행 기록은 lineage보다 먼저 전송하므로, lineage 수집 결과와 관계없이 항상 기록됩니다.

provider 1.7.0, 2.11.0, 2.20.2에서 같은 lineage가 저장되는지 비교한 결과는 tests/compat/README.md에 있습니다.

공식 플러그인과의 차이

URN 생성 규칙(DataFlow, DataJob, Dataset, DataProcessInstance)은 공식 플러그인과 같습니다. 따라서 공식 플러그인에서 전환해도 엔티티가 중복 생성되지 않습니다. tests/unit/test_sdk_parity.py가 이를 검증합니다.

개발

pip install -e ".[dev]"
pytest

test_sdk_parity.py는 acryl-datahub가 설치된 경우에만 실행됩니다. test_listener.py는 설치된 Airflow의 실제 listener manager(pluggy)에 등록해 훅 시그니처 호환성을 검증합니다.